#include<stdio.h>
#include <boost/pointer_cast.hpp>
#include "OrderBaseNode.hpp"

using namespace BFT::order;

OrderBaseNode::OrderBaseNode(std::string _membership, int _myId)
{
	if (BFT::order::Parameters::maxPeriods != 2)
	{
		fprintf(stderr, "maxPeriods must be 2\n");
		exit(110);
	}
	
	membership.reset(new BFT::membership::Membership(_membership, BFT::util::ORDER, _myId));
	baseNode.reset(new BFT::BaseNode(BFT::util::ORDER, membership));
	
	
	//init variables
	count = 0;
	procFirstReq = false;
	procFirstFiltReq = false;
	filtersmall = 0;
	filtermed = 0;
	filterlarge = 0;	

	deltaOrdered=0;
	previousOrdered=0;
	nowTime=0;
	previousTime=0;
	deltaTime=0;
	speed = 0; 
	
	nonDetcounter = millisecondTime();
	
	//TODO initialize filteredQuorums
	
	//init certificates
	certificates.resize(BFT::order::Parameters::maxPeriods);
	for (int i =0; i<BFT::order::Parameters::maxPeriods; i++)
		certificates.at(i).resize(BFT::order::Parameters::checkPointInterval);
	
	for(unsigned int i=0; i< certificates.size(); i++)
	{
		for (unsigned int j=0; j< certificates.at(i).size(); j++)
		{
			bool cp = ( j == certificates.at(i).size() -1 );
			certificates.at(i).at(j).reset(
				new BFT::order::statemanagement::Certificate(BFT::Parameters::getOrderCount(), cp) );
		}
	}
	
	//init futurePPs
	futurePPs.resize(BFT::order::Parameters::checkPointInterval);
	
	//init futurePs and futureCs
	futurePs.resize(BFT::order::Parameters::checkPointInterval);
	futureCs.resize(BFT::order::Parameters::checkPointInterval);
	for(unsigned int i = 0; i< futurePs.size(); i++)
	{
		futurePs.at(i).resize(BFT::Parameters::getOrderCount());
		futureCs.at(i).resize(BFT::Parameters::getOrderCount());

	}
	
	//init statesnapshots
	stateSnapshots.resize(BFT::order::Parameters::maxPeriods);
	stateSnapshots.at(0).reset(
		new BFT::order::statemanagement::CheckPointState(BFT::Parameters::getNumberOfClients()) );
	
	//fprintf(stdout, "order base node constructor, number of clients: %d\n",BFT::Parameters::getNumberOfClients());
	
	//init workingState
	workingState.reset( 
		new BFT::order::statemanagement::CheckPointState(stateSnapshots.at(0)) );
		
	currentIndex=0;
	baseIndex=0;
	
	//init cpCert
	cpCert.reset( new BFT::messages::Quorum<BFT::messages::CPTokenMessage>(
				BFT::Parameters::getExecutionCount(),
				BFT::Parameters::largeExecutionQuorumSize(),
				0) );

	//init retransmission backoffs
	lastLastExecCheck.resize(BFT::Parameters::getExecutionCount());
	lastExecDelay.resize(BFT::Parameters::getExecutionCount());
	lastExecLoadCP.resize(BFT::Parameters::getExecutionCount());	
	for(unsigned int i=0; i< lastExecLoadCP.size(); i++)
	{
		lastLastExecCheck.at(i) = 0;
		lastExecDelay.at(i) = 100;
		lastExecLoadCP.at(i) = false;
	}
	
	lastClientRetrans.resize(BFT::Parameters::getNumberOfClients());
	lastRidForClient.resize(BFT::Parameters::getNumberOfClients());
	for (int i=0; i< BFT::Parameters::getNumberOfClients();i++)
		lastRidForClient.at(i)=0;


	//init viewchange stuff	
	changingView = false;
	viewChanges.resize(BFT::Parameters::getOrderCount());
	for (int i=0; i < BFT::Parameters::getOrderCount(); i++)
	{	
		viewChanges.at(i).reset( new statemanagement::ViewChangeRow() );
	}
	nvCache.resize(BFT::Parameters::getOrderCount());
	actedOnView = false;
	actedOnDefinedView = false;
	
	currentLog = 0;
	
	//TODO: init the threadpool
	
	//obn = this; TODO needed??? 
	
	maxPrepared = -1;
	maxCommitted = -1;
	CPNOW = 27;
	requestsOrdered = 0;	
	

}

void OrderBaseNode::setNetwork(boost::shared_ptr<BFT::network::concurrentNet::TCPNetwork> _net)
{
	switch (_net.get()->getMyRole())
	{
		case BFT::util::CLIENT:
//			fprintf(stderr, "setting client network\n");
			clientNet = _net;
			break;
		case BFT::util::ORDER:
//			fprintf(stderr, "setting order network\n");
			orderNet = _net;
			break;
		case BFT::util::EXEC:
//			fprintf(stderr, "setting exec network\n");
			execNet = _net;
			break;
		case BFT::util::FILTER:
//			fprintf(stderr, "setting filter network\n");
			filterNet = _net;
			break;
		default:
			fprintf(stderr, "unknown network!!!\n");
			exit(120);		
	}
}


void OrderBaseNode::handle(boost::shared_ptr<char> _bytes, int _bytes_size)
{
//	usleep(1000);
//	boost::shared_ptr<BFT::messages::VerifiedMessageBase> vmb (
//		new BFT::messages::VerifiedMessageBase(_bytes, _bytes_size));
	
//	switch(vmb.get()->getTag())

//	if(baseNode.get()->getMyOrderIndex() == 3)
//	{
//		fprintf(stderr, "plainly ignoring all the messages...\n");
//		return;
//	}

	switch(ntohs( *(uint16_t*)_bytes.get() ))
	{
		case CLIENT_REQUEST:
			{
	
//				fprintf(stderr, "received a client request \t%d\n",millisecondTime());			
//				fprintf(stderr, "received a client request 1 , size %d\n",_bytes_size);		
//				for (int i=0; i< _bytes_size; i++)
//					fprintf(stdout,"|%02X",(unsigned char)_bytes.get()[i]);
//				fprintf(stdout,"\n");	
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::messages::ClientRequest> cr (
					new BFT::messages::ClientRequest(_bytes, _bytes_size));
//				fprintf(stderr, "received a client request 2 , size %d\n",_bytes_size);			
				process(cr);
			}
			break;
		//TODO filteredrequest
		case LAST_EXECUTED:
			{
//				fprintf(stderr, "received a last executed\n");			
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::messages::LastExecuted> le (
					new BFT::messages::LastExecuted(_bytes, _bytes_size));
				process(le);
			}
			break;
		case CP_LOADED:	
			{
				fprintf(stderr, "received a cp loaded\n");			
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::messages::CPLoaded> cpl (
					new BFT::messages::CPLoaded(_bytes, _bytes_size));
				process(cpl);
			}
			break;
		case CP_TOKEN_MESSAGE:
			{
//				fprintf(stderr, "received a cp token\n");			
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::messages::CPTokenMessage> cpt (
					new BFT::messages::CPTokenMessage(_bytes, _bytes_size));
				process(cpt);	
			}
			break;
		case PRE_PREPARE:
			{
//				TODO: find a way to cast without parsing everything again?	
//				fprintf(stderr, "received a dirty preprepare!!! \t%d\n",millisecondTime());			
				boost::shared_ptr<BFT::order::messages::PrePrepare> pp (
					new BFT::order::messages::PrePrepare(_bytes, _bytes_size));
				process(pp);
			}	
			break;
		case PREPARE:
			{
//				fprintf(stderr, "received a prepare\n");			
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::order::messages::Prepare> p (
					new BFT::order::messages::Prepare(_bytes, _bytes_size));
				process(p);
			}
			break;
		case COMMIT:
			{
//				fprintf(stderr, "received a commit\n");			
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::order::messages::Commit> c (
					new BFT::order::messages::Commit(_bytes, _bytes_size));
				process(c);
			}
			break;
		case VIEW_CHANGE:
			{
//				fprintf(stderr, "received a View Change\n");
				boost::shared_ptr<BFT::order::messages::ViewChange> vc(
					new BFT::order::messages::ViewChange(_bytes, _bytes_size));
				process(vc);
			}
			break;	
		case VIEW_CHANGE_ACK:
			{
//				fprintf(stderr, "received a view change ack - NIY!\n");
				boost::shared_ptr<BFT::order::messages::ViewChangeAck> vca(
					new BFT::order::messages::ViewChangeAck(_bytes, _bytes_size));
				process(vca);
			}
			break;
		case NEW_VIEW:
			{
//				fprintf(stderr, "received a new view - NIY!\n");
				boost::shared_ptr<BFT::order::messages::NewView> nv(
					new BFT::order::messages::NewView(_bytes, _bytes_size));
				process(nv);
			}
			break;
		case MISSING_VIEW_CHANGE:
			{
				fprintf(stderr, "received a missing view change - NIY!\n");
			}
			break;
		case CONFIRM_VIEW:
			{
				boost::shared_ptr<BFT::order::messages::ConfirmView> cv(
					new BFT::order::messages::ConfirmView(_bytes, _bytes_size));
				process(cv);
			}
			break;
		case RELAY_VIEW_CHANGE:
			{
				fprintf(stderr, "received a relay view change - NIY!\n");
			}
			break;
		case MISSING_OPS:
			{
				fprintf(stderr, "received a missing ops - NIY!\n");
			}
			break;
		case RELAY_OPS:
			{
				fprintf(stderr, "received a relay ops - NIY!\n");
			}
			break;
		case START_VIEW:
			{
//				fprintf(stderr, "received a start view - NIY!\n");
				boost::shared_ptr<BFT::order::messages::StartView> sv(
					new BFT::order::messages::StartView(_bytes, _bytes_size));
				process(sv);
			}
			break;
		case MISSING_CP:
			{
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::order::messages::MissingCP> mcp (
					new BFT::order::messages::MissingCP(_bytes, _bytes_size));
			
				fprintf(stderr, "received a missing cp from %d\n",mcp.get()->getSender());			
				process(mcp);
			}
			break;
		case RELAY_CP:
			{
				fprintf(stderr, "received a relay cp\n");			
				//TODO: find a way to cast without parsing everything again?				
				boost::shared_ptr<BFT::order::messages::RelayCP> rcp (
					new BFT::order::messages::RelayCP(_bytes, _bytes_size));
				process(rcp);
			}
			break;
		default:
			fprintf(stderr,"Order does not know how to handle message %d\n", *(uint16_t*)_bytes.get() );
			exit(110);
	}
}

void OrderBaseNode::handle(boost::shared_ptr<messages::PrePrepare> _pp)
{
//	fprintf(stderr,"Handling a clean preprepare message!!! \t%d\n",millisecondTime());
	processClean(_pp);
}

void OrderBaseNode::handle(boost::shared_ptr<std::vector<boost::shared_ptr<BFT::messages::RequestCore> > > _rcs)
{
	//fprintf(stderr,"vector of RequestCores handling is not implemented!!!");	
	processClean(_rcs);
}


void OrderBaseNode::process(boost::shared_ptr<BFT::messages::ClientRequest> _cr)
{
//	fprintf(stderr, "processing a cr, use count: %ld, ptr: %p\n",_cr.use_count(),(void*) _cr.get());
	if(!baseNode.get()->validateClientMacArrayMessage(_cr))
	{
		fprintf(stderr, "client request did not validate\n");
	}
	//fprintf(stderr, "processing the same cr, use count: %ld\n",_cr.use_count());	
	boost::shared_ptr<BFT::messages::RequestCore> rc = _cr.get()->getCore();
	
	if ( _cr.get()->getSender()!= rc.get()->getSendingClient() )
	{
		fprintf(stderr,"should not get a client request from a different client than the core\n");
		exit(111);
	}
	
//	fprintf(stderr, "before tryretransmit\n");
	
	if (tryRetransmit(rc))
		return;
	
	cleaner.get()->clean(rc);
}


//resend nextbatch messages from last+1 through currentseqno -1
//Since the lastexecuted message is sent by a single replica
//(rather than a quorum), it must be rate limited.
void OrderBaseNode::process(boost::shared_ptr<BFT::messages::LastExecuted> _le)
{
//	fprintf(stderr, "*** Processing a LastExecuted message ***\n");

	uint32_t cur = millisecondTime();
	uint32_t senderIndex = _le.get()->getSendingReplica();
	if (cur < lastLastExecCheck.at(senderIndex) + lastExecDelay.at(senderIndex))
	{
//		fprintf(stderr, "orderbasenode::process(lastexecuted) says: LE ignored due to rate limiting\n");
		return;
	}
	else
	{
		lastLastExecCheck.at(senderIndex) = cur;
		lastExecDelay.at(senderIndex) *= 2;
		if(lastExecDelay.at(senderIndex) > 1000)
		{
			lastExecDelay.at(senderIndex) = 300;
		}
	}
	
	uint32_t baseline = _le.get()->getSequenceNumber();
	if(baseline > getCurrentSequenceNumber())
	{
		fprintf(stderr,"OrderBasENode::process(lastexecuted) says: ignoring LE, baseline %d > current seqno %d!\n",
			baseline, getCurrentSequenceNumber());
		return;
	}
	
	if(! baseNode.get()->validateExecMacArrayMessage(_le))
	{
		fprintf(stderr,"OrderBasENode::process(lastexecuted) says: LE has invalid mac array!\n");
		return;
	}
	
	// if it is too old, send a load checkpoint
	if(baseline < getBaseSequenceNumber())
	{
		int token_bytes_size;
		boost::shared_ptr<char> token_bytes = stateSnapshots.at(baseIndex).get()->getExecCPToken(token_bytes_size);
//		fprintf(stderr,"orderbasenode::process(lastexecuted) says : token_bytes_size %d\n",token_bytes_size);
		boost::shared_ptr<BFT::messages::LoadCPMessage> lcp (
			new BFT::messages::LoadCPMessage( token_bytes, token_bytes_size,
							  getBaseSequenceNumber(),
							  baseNode.get()->getMyOrderIndex()  )); 
							  
		baseNode.get()->authenticateExecMacMessage(lcp, (int)senderIndex);
		baseline = getBaseSequenceNumber();
		int lcp_bytes_size;
		boost::shared_ptr<char> lcp_bytes = lcp.get()->getBytes(lcp_bytes_size);
//		//verifying the message correctness
//		boost::shared_ptr<BFT::messages::LoadCPMessage> new_lcp (
//			new BFT::messages::LoadCPMessage(lcp_bytes,lcp_bytes_size));
//		new_lcp.get()->printInfo();
//		fprintf(stderr,"orderbasenode::process(lastexecuted) says : sending a load checkpoint\n");
		baseNode.get()->sendToExec(lcp_bytes, lcp_bytes_size, senderIndex);
		lastExecLoadCP.at(senderIndex) = true;
		return;
	}
//	fprintf(stderr,"orderbasenode::process(lastexecuted) says : before forwarding\n");	
	forwardNextBatches(baseline, senderIndex);
//	fprintf(stderr,"orderbasenode::process(lastexecuted) says : after forwarding\n");	
}

void OrderBaseNode::forwardNextBatches(uint32_t _baseline, uint32_t _sendingReplica)
{
	int count =0;
	int nb_bytes_size;
	uint32_t baseline = _baseline;
	boost::shared_ptr<char> nb_bytes;
	//if is in range, send the next batches up to the current sequence number
	if(baseline < getCurrentSequenceNumber())
	{
		int index = stateIndex(_baseline);
		while (index != currentIndex)
		{
			for(unsigned int i = certificateIndex(baseline); 
				i < certificates.at(index).size();
				i++, baseline++)
			{
				boost::shared_ptr<BFT::messages::NextBatch> nb = 
					certificates.at(index).at(i).get()->getNextBatch(baseNode,currentView);
				count++;
				baseNode.get()->authenticateExecMacArrayMessage(nb);
				nb_bytes = nb.get()->getBytes(nb_bytes_size);
				baseNode.get()->sendToExec(nb_bytes, nb_bytes_size, _sendingReplica);	
			}
			index = (index+1) % BFT::order::Parameters::maxPeriods;	
		}
		
		//we are now in the working certificate list
		for(int i= certificateIndex(baseline); baseline < getCurrentSequenceNumber(); i++)
		{
			boost::shared_ptr<BFT::messages::NextBatch> nb =	
				certificates.at(index).at(i).get()->getNextBatch(baseNode, currentView);
			baseNode.get()->authenticateExecMacArrayMessage(nb);
			count++;
			nb_bytes = nb.get()->getBytes(nb_bytes_size);
			baseNode.get()->sendToExec(nb_bytes, nb_bytes_size, _sendingReplica);
			baseline++;
		}
	}
	
//	fprintf(stderr, "OrderBaseNode::forwardNextBatches says: sent %d messages starting at %d in response to a last executed\n",
//		count, baseline);
}


void OrderBaseNode::process(boost::shared_ptr<BFT::messages::CPLoaded> _cpl)
{
	fprintf(stderr, "processing a CPLoadedMessage \n");
	int senderIndex = _cpl.get()->getSendingReplica();
	if(!lastExecLoadCP.at(senderIndex))
		return;
	
	uint32_t baseline = _cpl.get()->getSequenceNumber();
	if(baseline > getCurrentSequenceNumber())
		return;
	if(baseline < getBaseSequenceNumber())
		return;
	if(!baseNode.get()->validateExecMacArrayMessage(_cpl))
		return;
	forwardNextBatches(baseline, senderIndex);
	lastExecLoadCP.at(senderIndex)=false;	
}

void OrderBaseNode::process(boost::shared_ptr<BFT::messages::CPTokenMessage> _cpt)
{
	fprintf(stderr, "processing cptoken %d\n",_cpt.get()->getSequenceNumber());
	
	uint32_t token_seqno = _cpt.get()->getSequenceNumber();
	
	//discard old tokens
	if ( token_seqno <= getBaseSequenceNumber() &&
	     (token_seqno != 0  || getBaseSequenceNumber() != 0 ) )
	{
		fprintf(stderr,"token is old, discarded.\n");
		return;
	}
	
	//validate the message
	if ( !baseNode.get()->validateExecMacArrayMessage(_cpt))
	{
		fprintf(stderr, "invalid CPToken!\n");
		exit(119);
	}
	
	//add to certificate
	cpCert.get()->addEntry(_cpt);
//	fprintf(stderr,"after addEntry\n");
	//act on it, if it is complete
	if (cpCert.get()->isComplete())
	{
		fprintf(stderr,"acting on cptoken %d\n",_cpt.get()->getSequenceNumber());
		actonCPToken(_cpt);
		//fprintf(stderr,"after acton\n");
	}
	else
	{
		fprintf(stderr,"{{{{}}}} = not complete, not acting on cptoken %d = {{{{{}}}}}\n",_cpt.get()->getSequenceNumber());
	}
	
}


//act upon a complete CPToken
void OrderBaseNode::actonCPToken(boost::shared_ptr<BFT::messages::CPTokenMessage>)
{
//	fprintf(stderr, "inside acton for a cptoken");
	if (!cpCert.get()->isComplete())
	{
		fprintf(stderr, "orderbasenode::acton(cpcert) says : pccert has to be complete!!!!!\n");
		exit(120);
	}
//	else
//	{
//		fprintf(stderr, "cptoken is complete, going on...\n");
//	}
	
	boost::shared_ptr<BFT::messages::CPTokenMessage> cptok = cpCert.get()->getEntry();
	
	if (cptok.get()->getSequenceNumber() <= getBaseSequenceNumber() &&
	    cptok.get()->getSequenceNumber() != 0 ) //triggers during startup... and hopefully ONLY during startup
	{
		fprintf(stderr, "complete cpCert has a cptoken entry coming from the distant past...\n");
		exit(121);
	}
//	else
//	{
//		fprintf(stderr, "cptoken not from the past, going on...\n");
//	}
			
	if (!changingView)
	{
		uint32_t seqno = cptok.get()->getSequenceNumber();
		
		//get the check point state to which to add the cptok
		int state = stateIndex(seqno);
		boost::shared_ptr<statemanagement::CheckPointState> cps;
		if (workingState.get()->getBaseSequenceNumber() < seqno)
		{
			cps = workingState;
			//fprintf(stderr, "cps taken from working state...\n");
		}
		else
		{
			cps = stateSnapshots[state];
//			fprintf(stderr, "cps taken from snapshots...\n");
		}
		
		if ( seqno > (cps.get()->getBaseSequenceNumber() + 
					(BFT::order::Parameters::maxPeriods * 
					BFT::order::Parameters::checkPointInterval)) )
		{
			fprintf(stderr, "received a cptoken way off in the future\n");
			exit(121);
		}
		//else
		//{
		//	fprintf(stderr, "cptoken not too much in the future, going on...\n");		
		//}
		
		if ( cps.get()->getBaseSequenceNumber() + BFT::order::Parameters::checkPointInterval != seqno &&
			seqno !=0)
		{
			fprintf(stderr, "trying to add a seqno to a wrong state\n");
			exit(122);
		}
		//else
		//{
		//	fprintf(stderr, "cptoken in the right state...\n");
		//}
		
		if (cps.get()->hasExecCP())
		{
			fprintf(stderr,"cps has already a cptok\n");
			return;
		}
		
		int token_bytes_size;
		boost::shared_ptr<char> token_bytes = cptok.get()->getToken(token_bytes_size);

//		fprintf(stderr, "before addexeccptoken...\n");	
		cps.get()->addExecCPToken(token_bytes, token_bytes_size, cptok.get()->getSequenceNumber());
//		fprintf(stderr, "after addexeccptoken...\n");	
		
		//if the last seqno before the checkpoint is committed, mark the checkpoint as stable
		boost::shared_ptr<statemanagement::Certificate> cert;
		
		if(seqno > 0)
		{
//			fprintf(stderr, "seqno > 0...\n");	
			cert = certificates.at(stateIndex(seqno-1)).at(certificateIndex(seqno-1));
		}
		if(seqno == 0 || cert.get()->isCommitted())
		{
//			fprintf(stderr,"seqno == 0 || cert is committed ...\n");	
			makeCPStable(cps);
//			fprintf(stderr, "after makecpstable\n");	
			if(seqno == 0)
			{
//				fprintf(stderr, "before commit...\n");	
				cps.get()->commit();
//				fprintf(stderr, "after commit...\n");	
			}
		}
		else
		{
			fprintf(stderr, "not making stable a cptok with seqno:%d\n",seqno);
		}

	}
	else
	{
		cachedCPT = cptok;
	}
	cpCert.get()->clear();	
}

void OrderBaseNode::makeCPStable(boost::shared_ptr<statemanagement::CheckPointState> _cps)
{
	if (_cps.get()->isStable() || _cps.get()->isMarkingStable())
	{
		fprintf(stderr, "cps alrady stable, getting out...\n");	
		return;
	}
	
	if (BFT::order::Parameters::doLogging)
	{
		//TODO implement logging stuff here
		fprintf(stderr, "trying to log...\n");	
	}
	else
	{
		//fprintf(stderr, "before marking stable...\n");	
		_cps.get()->markingStable();
		//fprintf(stderr, "before making stable...\n");	
		_cps.get()->makeStable();
		//fprintf(stderr, "before cps stable...\n");	
		cpStable(_cps);
	}
}

void OrderBaseNode::cpStable(boost::shared_ptr<statemanagement::CheckPointState> _cps)
{
	if (!_cps.get()->isStable())
	{
		fprintf(stderr, "orderbasenode::cpstable says : checkpoint has to be stable!!!\n");
		exit(122);
	}
	
	uint32_t seqno = _cps.get()->getCurrentSequenceNumber();
	int state = stateIndex(seqno);
	int cert = BFT::order::Parameters::checkPointInterval - 1;
	
	if (!changingView)
	{
		boost::shared_ptr<statemanagement::Certificate> certificate;
		certificate = certificates.at(state).at(cert);
		boost::shared_ptr<messages::PrePrepare> pp = certificate.get()->getCachedPrePrepare();
		if(pp.use_count() != 0)
		{
			process(pp);
		}
	}
	else
	{
		if (vInfo.get()->definedView())
		{
			actOnDefinedView();
		}
	    	else
	    	{
			// do nothing if the view isnt defined yet
		}
	}
}


//Process a PrePrepare.
//Check that the message is
//(1) part of this view
//otherwise discard it
//(2) sent by the primary
//otherwise discard it
//(3) the next request
//(a) if it is old then discard it
//(b) if it is in the 'near' future then cache it
//for later processing
//(c) otherwise discard it
//(4) cache it if we're waiting for the previous commit
//(5) check the MAC
//(6) the time/nondeterminism are consistent
//(7) The histories check out appropriately
//i.e. the base history is consistent with the history
//defined by the previous sequence number
//(8) it contains an appropriate order CP Digest
//(a)  last request in a cp interval has the 
//about to be commited CP associated with it.
//and that cp is stable
//(b)  other sequence numbers have the 
//last committed sequence number
//and that cp is stable
//(c)  if a cp is necessary but is not yet stable, 
//then cache the preprepare
//(9) Check the signatures on the request cores
void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::PrePrepare> _pp)
{


//	fprintf(stderr, "process a PrePrepare: %d/%d:%d view: %d\n",_pp.get()->getSeqNo(), getCurrentSequenceNumber(),
//		getBaseSequenceNumber(), _pp.get()->getView() );
	
	//1 discard if not part of this view
	if(_pp.get()->getView() != currentView)
	{
//		fprintf(stderr, "OrderBaseNode::process(preprepare) says : discarding out of view preprepare\n");
		return;
	}
	
	//2 drop the message if it is not from the primary
	if ((int)_pp.get()->getSendingReplica() != primary(currentView) )
	{
		fprintf(stderr, "OrderBaseNode::process(preprepare) says : discarding preprepare not from the primary\n");
		return;
	}
	
	uint32_t seqno = _pp.get()->getSeqNo();

	//3a drop if too old
	if (seqno < getCurrentSequenceNumber())
	{
		fprintf(stderr, "OrderBaseNode::process(preprepare) says : discarding old preprepare\n");
		return;
	}
	
	if( seqno >= (getBaseSequenceNumber() + Parameters::maxPeriods*Parameters::checkPointInterval) )
	{
		//3b preprepare coming from the future. if it is not too far from the future, cache it
		if(seqno < (getBaseSequenceNumber() + (Parameters::maxPeriods + 1)*Parameters::checkPointInterval))
		{
			futurePPs.at(seqno % Parameters::checkPointInterval) = _pp;
			return;
		}
		else //3c too far in the future, drop it
		{
			fprintf(stderr, "OrderBaseNode::process(preprepare) says : discarding preprepare from the far future\n");
			return;
		}
	}
	
	boost::shared_ptr<statemanagement::Certificate> cert = certificates.at(stateIndex(seqno)).at(certificateIndex(seqno));
	
	if( seqno != getCurrentSequenceNumber() || changingView ) //pp in the near future, cache it
	{
		cert.get()->cachePrePrepare(_pp);
		return;
	}
	
	//4 cache the message if we are waiting for a commit
	if (seqno % Parameters::checkPointInterval == 0
		&& seqno > 0
		&& !certificates.at(stateIndex(seqno-1)).at(certificateIndex(seqno-1)).get()->isCommitted() )
	{
		cert.get()->cachePrePrepare(_pp);
		return;
	}
	
	//5 check the MAC
	if(!baseNode.get()->validateOrderMacArrayMessage(_pp) ||
	   !_pp.get()->checkAuthenticationDigest())
	{
		fprintf(stderr, "OrderBaseNode::process(preprepare) says : MAC array did not validate\n");
		return;
	}
	
//	_pp->printInfo();
	
	//6 check that nondet time is consistent
	uint32_t time = _pp.get()->getNonDeterminism().get()->getTime();
	uint32_t current = millisecondTime();
	uint32_t lasttime = workingState.get()->getCurrentTime();
	if(/*(lasttime !=0 && time < lasttime) ||*/ 
	    ( current < time && (time - current) > Parameters::timeVariance ) ||
	    ( time < current && (current - time) > Parameters::timeVariance )  )
	{
		fprintf(stderr, "OrderBaseNode::process(preprepare) says : time check failed\n");
		fprintf(stderr, "time %ld, lasttime %ld, current %ld, variance %ld\n",time,lasttime,current,Parameters::timeVariance);
		return;
	}
	
	//7 check that the initial history matches the final history of the preceding sequence number
	boost::shared_ptr<BFT::messages::HistoryDigest> hist;
	if( stateIndex(seqno) == baseIndex && certificateIndex(seqno) == 0)
	{
		hist = stateSnapshots.at(baseIndex).get()->getHistory();
	}
	else
	{
		hist = certificates.at(stateIndex(seqno-1)).at(certificateIndex(seqno-1)).get()->getHistory();
	}
	
	if(! _pp.get()->getHistory().get()->equals(hist) )
	{
		fprintf(stderr, "OrderBaseNode::process(preprepare) says : dropping a preprepare, messed up with histories\n");
		return;
	}
	
	//8 check that the checkpoint snapshots match
	bool cp = ((getCurrentSequenceNumber() + 1) % Parameters::checkPointInterval) == 0;
	//8b  not the last sequencenumber matches with the base cp index
	if(!cp &&
	   !_pp.get()->getCPHash().get()->equals( stateSnapshots.at(baseIndex).get()->getStableDigest() ))
	{
		fprintf(stderr, "OrderBaseNode::process(preprepare) says : dropping a preprepare, checkpoints do not match (1)\n");
		return;	
	}
	
	//8a the last sequence number in a CP interval matches the next hash
	int baseplusone = (baseIndex+1) % Parameters::maxPeriods;
	if(baseIndex != currentIndex && cp)
	{
		//8c if the checkpoint is not stable, return
		if( !stateSnapshots.at(baseplusone).get()->isStable() )
		{
			cert.get()->cachePrePrepare(_pp);
			return;
		}
		
		//8a hashes have to match
		if( !_pp.get()->getCPHash().get()->equals( stateSnapshots.at(baseplusone).get()->getStableDigest()) )
		{
			fprintf(stderr, "OrderBaseNode::process(preprepare) says : dropping a preprepare, checkpoints do not match(2)\n");
			return;
		}
	}
	
	
	//9 (finally) check the signatures on the request cores
	if (! BFT::Parameters::filtered )
	{
		cleaner.get()->clean(_pp);
	}
	else
	{
		//TODO implement filtered related stuff
//		RequestCore[] reqs = pp.getRequestBatch().getEntries();
//		for (int i = 0; i < reqs.length; i++)
//			if (!validateFullFilterMacSignatureMessage((FilteredRequestCore)reqs[i],
//					BFT.Parameters.mediumFilterQuorumSize()))
//				Debug.kill("failed verification");
//		if (pp.checkAuthenticationDigest())
//		    processClean(pp);
//		else
//		    BFT.Debug.kill("bad things!");
		
		fprintf(stderr, "OrderBaseNode::process(preprepare) says : FILTERED REQUESTS NIY!!!\n");
		exit(151);
	}
}

/**
Process a prepare messages

(1)  Check that it is for the current view
(3)  discard it if the sequence number is old
(4)  if its for the mid range future then cache it
(5)  if it is already committed then discard
(6)  if it does not match the preprepare then discard
(7)  validat ethe authentication
(8) if validating local state prepares, then act on it
 **/
void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::Prepare> _p)
{
	
	//1 check if it is in the current view
	if (_p.get()->getView() != currentView || changingView)
	{
//		fprintf(stderr,"orderbasenode::process (prepare) says : ignoring prepare, out of view or viewchanging\n");
		return;
	}
	
	uint32_t seqno = _p.get()->getSeqNo();
	uint32_t sender = _p.get()->getSendingReplica();
//	fprintf(stderr, "@\tprocessing a Prepare for seqno %d from %d\n",seqno,sender);
	//3 discard if the seqno is old
	if(seqno < getBaseSequenceNumber())
	{
		fprintf(stderr,"orderbasenode::process (prepare) says : ignoring prepare, old seqno\n");
		return;
	}
	
	//4 cache if it is not the next
	if(seqno >= getBaseSequenceNumber() +
		Parameters::maxPeriods*Parameters::checkPointInterval)
	{
		if(seqno < getBaseSequenceNumber() +
			(Parameters::maxPeriods+1)*Parameters::checkPointInterval)
		{
			int ind = (seqno % Parameters::checkPointInterval);
			fprintf(stderr,"orderbasenode::process(prepare) says: caching prepare in the near future\n");
			futurePs.at(ind).at(sender) = _p;
			return;
		}
		else //too distant in the future, discarded
		{
			fprintf(stderr,"orderbasenode::process(prepare) says: ignoring prepare in the distant future\n");
			return;
		}
	}
	
	//caching not yet ready
	boost::shared_ptr<statemanagement::Certificate> cert = 
		certificates.at(stateIndex(seqno)).at(certificateIndex(seqno));
	
	if (!cert.get()->isPrePrepared())
	{
		fprintf(stderr, "@\tcert not complete yet, caching Prepare %d %d\n",seqno,sender);
		cert.get()->cachePrepare(_p);
		return;
	}
	
	if(cert.get()->preparedBy(sender) && (int)sender != primary(currentView))
	{
		fprintf(stderr, "@\tcert already prepared by a !primary, dropping prepare %d %d\n",seqno,sender);
		return;
	}
	
	//5 if it is already committed then discard it
	if (cert.get()->isCommitted() &&
	    (!cert.get()->preparedBy(sender) && ((int)sender == primary(currentView) || amIPrimary(currentView)) ) )
	{
		fprintf(stderr, "@\tcert already committed, dropping prepare %d %d\n",seqno,sender);
		return;
	}
	
	//6 check that it matches the preprepare
	if(!_p.get()->getPrePrepareHash().get()->equals(cert.get()->getHistory() ) )
	{
		fprintf(stderr,"orderbasenode::processprepare says : histories do not match\n");
		exit(153);
	}
	
	//7 validate authentication
	if(!baseNode.get()->validateOrderMacArrayMessage(_p))
	{
		fprintf(stderr,"orderbasenode::processprepare says : prepare did not validate\n");
		exit(153);	
	} 
	
	//7a if it came from the primary, send a prepare
	bool wasPrepped = cert.get()->isPrepared();
	bool added1 = false;
	if((int)sender == primary(currentView) &&
	   (int)sender != baseNode.get()->getMyOrderIndex())
	{
		if (cert.get()->getPrepare().use_count()==0  )
		{
			fprintf(stderr,"orderbasenode::processprepare says : prepare %d from primary, mee too!!!\n",
				_p.get()->getSeqNo() );
			boost::shared_ptr<messages::Prepare> p2 (
				new messages::Prepare(_p.get()->getView(), _p.get()->getSeqNo(),
						      _p.get()->getPrePrepareHash(), 
						      baseNode.get()->getMyOrderIndex()) );
			baseNode.get()->authenticateOrderMacArrayMessage(p2);
			int bytes_size;
			boost::shared_ptr<char> bytes = p2.get()->getBytes(bytes_size);
			baseNode.get()->sendToAllOrder(bytes,bytes_size);
			added1 = cert.get()->addMyPrepare(p2); ////////////////////////////
//			if (added1)
//				fprintf(stderr, "@\tadding my prepare %d - %d\n",seqno,sender);
//			else	
//				fprintf(stderr, "@\tUNABLE TO ADD my prepare %d - %d\n",seqno,sender);

		}
	}
	bool added2 = false;
	if((int)sender == baseNode.get()->getMyOrderIndex())
	{
		added2 = cert.get()->addMyPrepare(_p); 
//		if (added2)
//			fprintf(stderr, "@\tadding my prepare %d - %d\n",seqno,sender);
//		else	
//			fprintf(stderr, "@\tUNABLE TO ADD my prepare %d - %d\n",seqno,sender);
	}
	else
	{   
		added2 = cert.get()->addOtherPrepare(_p);
//		if(added2)
//			fprintf(stderr, "@\tadding other prepare %d - %d\n",seqno,sender);
//		else
//			fprintf(stderr, "@\tUNABLE TO ADD other prepare %d - %d\n",seqno,sender);		
	}
		
	//8 add it on local state. if that prepares, act on it
	if(/*(added1 || added2) &&*/ !wasPrepped && cert.get()->isPrepared())
	{	
//		fprintf(stderr, "@\tcert became prepared!\n");
		actOnPrepared(_p.get()->getSeqNo());	
	}
}

//Process a commit

//(1)  check that it is for the current view
//(2)  discard old sequence numbers
//(3)  cache mid range future requests
//(4)  discard if it is alrady committed
//(5)  discard if it does not match the preprepare
//(6)  validate authentication
//(7)  if it makes the certificate complete, then act on it
void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::Commit> _c)
{
	if (_c.get()->getView() != currentView || changingView)
	{
//		fprintf(stderr,"orderbasenode::process (commit) says : ignoring commit, out of view or viewchanging\n");
		return;
	}
	
	uint32_t seqno = _c.get()->getSeqNo();
	uint32_t sender = _c.get()->getSendingReplica();
	
	fprintf(stderr,"@\t\tprocessing a Commit for seqno %d - sender %d\n",seqno,sender);
	
	//2 discard old requests
	if(seqno < getBaseSequenceNumber())
	{
		fprintf(stderr,"orderbasenode::process (commit) says : ignoring commit, too old\n");
		return;
	}
	
	//4 cache if it is not the next
	if(seqno >= getBaseSequenceNumber() +
		Parameters::maxPeriods*Parameters::checkPointInterval)
	{
		if(seqno < getBaseSequenceNumber() +
			(Parameters::maxPeriods+1)*Parameters::checkPointInterval)
		{
			int ind = (seqno % Parameters::checkPointInterval);
			futureCs.at(ind).at(sender) = _c;
			return;
		}
		else //too distant in the future, discarded
		{
			fprintf(stderr,"orderbasenode::process(commit) says: ignoring commit in the distant future\n");
			return;
		}
	}
		//caching not yet ready
	boost::shared_ptr<statemanagement::Certificate> cert = 
		certificates.at(stateIndex(seqno)).at(certificateIndex(seqno));
	
	if (!cert.get()->isPrePrepared())
	{
		cert.get()->cacheCommit(_c);
		return;
	}
	
	//5 if it is already committed then discard it
	if (cert.get()->isCommitted())
	{
//		fprintf(stderr,"orderbasenode::process(commit) says: ignoring commit, cert already committed\n");
		return;
	}
	
	//6 check that it matches the preprepare
	if(!_c.get()->getPrePrepareHash().get()->equals(cert.get()->getHistory() ) )
	{
		fprintf(stderr,"orderbasenode::process(commit) says : histories do not match\n");
		exit(153);
	}
	
	//7 validate authentication
	if(!baseNode.get()->validateOrderMacArrayMessage(_c))
	{
		fprintf(stderr,"orderbasenode::process(commit) says : prepare did not validate\n");
		exit(153);	
	}
	
	// if it completes the certificate, act on it 
	bool wasPrepared = cert.get()->isPrepared();
	bool added = cert.get()->addOtherCommit(_c);
	if(added && cert.get()->isCommitted())
	{
		actOnCommitted(seqno);
	}
	else
	{
		if(added && !wasPrepared && cert.get()->isPrepared())
		{
			actOnPrepared(seqno);	
		}
	}
}


/**
*  Start a new view change.
*
*  (1) change the current view
*  (2) set the inviewchange flag
*  (3) create a view change message and send it to the world
**/
void OrderBaseNode::startViewChange(uint32_t _newView)
{

	if(changingView)
	{
		fprintf(stderr,"already viewchanging \n");
		exit(184);	
	}	
	
	fprintf(stderr,"OrderBaseNode::startViewChange says: Starting a View Change! new_view: %d!\n",_newView);
	if(currentView >= _newView)
	{
		fprintf(stderr, "OrderBaseNode::startViewChange says : View number should be monotonically increasing!\n");
		exit(162);
	}
	
	currentView = _newView;
	actedOnView = false;
	actedOnDefinedView = false;
	std::vector<boost::shared_ptr<statemanagement::ViewChangeCertificate> > vcCerts;
	vcCerts.resize(viewChanges.size());
	
	//check the cache of view change messages to see if we want to be in a larger view
	bool lookForLargerView = true;
	while(lookForLargerView)
	{
		//get the current view change
		for(unsigned int i=0; i<viewChanges.size(); i++)
		{
			vcCerts.at(i)=viewChanges.at(i).get()->getCertificate(currentView);
		}
		int count = 0;
		uint32_t small = currentView;
		
		//look to see if sufficient future view changes exist
		for(unsigned int i=0; i<viewChanges.size(); i++)
		{
			if(viewChanges.at(i).get()->getSmallestView() > currentView)
			{
				count++;
			}
			if(viewChanges.at(i).get()->getSmallestView() < small || small == currentView)
			{
				small = viewChanges.at(i).get()->getSmallestView();
			}
		}
		
		//if there is a future valid view, OR
		//if the primary has a larger view
		if(count >= BFT::Parameters::smallOrderQuorumSize() ||
		   viewChanges.at(primary(small)).get()->getLastReceived() > small )
		{
			//switch and look again
			currentView = small;
		}
		else
		{
			//otherwise stop looking
			lookForLargerView = false;
		}
	}
	
//	fprintf(stderr,"*********** looked for larger views\n");
	
	//establish the necessary information for the ViewChange message	
	uint32_t cpSeqno = getBaseSequenceNumber(); //highest committed CP seqno
	uint32_t stableCPSeqno = cpSeqno; //highest stable cp seqno
	uint32_t prepareSeqno = cpSeqno; //last prepared + 1
	uint32_t preprepareSeqno = cpSeqno; //last preprepared + 1
	
	//TODO STOP CHEATING!!!
	boost::shared_ptr<statemanagement::CheckPointState> committedCP;
	boost::shared_ptr<statemanagement::CheckPointState> stableCP;
	std::vector<boost::shared_ptr<BFT::messages::CertificateEntry> > entries;
	std::vector<boost::shared_ptr<BFT::messages::HistoryDigest> > histories;
	
	//compute cpSeqno
	uint32_t cpIndex = baseIndex;
	uint32_t tmp = (cpIndex+1)%stateSnapshots.size();
	while((int)tmp != baseIndex && stateSnapshots.at(tmp).get()->isCommitted())
	{
		cpIndex = tmp;
		tmp = (cpIndex+1)%stateSnapshots.size();
	}

	cpSeqno=stateSnapshots.at(cpIndex).get()->getCurrentSequenceNumber();
//	fprintf(stderr," *** 1 *** cpSeqno: %d, cpIndex: %d\n",cpSeqno, cpIndex);

	//compute StableCPSeqno
	stableCPSeqno = cpSeqno;
	while((int)tmp != baseIndex && stateSnapshots.at(tmp).get()->isStable())
	{
		cpIndex = tmp;
		tmp = (cpIndex+1)%stateSnapshots.size();
	}
	stableCPSeqno = stateSnapshots.at(cpIndex).get()->getCurrentSequenceNumber();
//	fprintf(stderr," *** 2 *** cpSeqno: %d, cpIndex: %d\n",cpSeqno, cpIndex);	
	//compute pSeqno
	while (prepareSeqno < getCurrentSequenceNumber() &&
		certificates.at(stateIndex(prepareSeqno)).at(certificateIndex(prepareSeqno)).get()->isPrepared() )
	{
		prepareSeqno++;
	}
	
	//compute preprepareSeqno
	preprepareSeqno = prepareSeqno;

//	fprintf(stderr," *** 3 *** cpSeqno: %d, cpIndex: %d\n",cpSeqno, cpIndex);	

	while ( preprepareSeqno < getCurrentSequenceNumber() &&
		certificates.at(stateIndex(preprepareSeqno)).at(certificateIndex(preprepareSeqno)).get()->isPrePrepared() )
	{	
		if ( certificates.at(stateIndex(preprepareSeqno)).at(certificateIndex(preprepareSeqno)).get()->isPrepared() )
		{
			fprintf(stderr,"OrderBaseNode::startViewChange says : there should not be prepared entries out of order\n");
			exit(163);
		}
		preprepareSeqno++;
	
	}
//	fprintf(stderr," *** 4 *** cpSeqno: %d, cpIndex: %d\n",cpSeqno, cpIndex);	
	if(preprepareSeqno != getCurrentSequenceNumber())
	{
		fprintf(stderr,"OrderBaseNode::startNewView says : so the current sequence number is *not* the last preprepared +1. \
				Suicide is the only meaningful thing to do. I die with honor.\n");
		exit(164);
	}
//	fprintf(stderr," *** 5 *** cpSeqno: %d, cpIndex: %d\n",cpSeqno, cpIndex);	
	//TODO stop cheating
	committedCP = stateSnapshots.at(stateIndex(cpSeqno));
	stableCP = stateSnapshots.at(stateIndex(stableCPSeqno));
//	fprintf(stderr," *** 6 *** cpSeqno: %d, cpIndex: %d\n",cpSeqno, cpIndex);	
	entries.resize(preprepareSeqno - cpSeqno);
	histories.resize(preprepareSeqno - cpSeqno);
	fprintf(stderr, "Starting a new view. preprepareSeqno %d, cpSeqno %d, entry size %d\n"
		, preprepareSeqno, cpSeqno, preprepareSeqno-cpSeqno);
		
//	fprintf(stderr, " --- copying %d entries ---\n", (int)entries.size() );
	
	int temp_int;
	boost::shared_ptr<char> temp_bytes;
	
	for(unsigned int i=0; i<entries.size(); i++)
	{
		long seq_no = cpSeqno + i;
		entries.at(i) = certificates.at(stateIndex(seq_no)).at(certificateIndex(seq_no)).get()->getCertEntry();
		temp_bytes = entries.at(i).get()->getBytes(temp_int); 
		histories.at(i) = certificates.at(stateIndex(seq_no)).at(certificateIndex(seq_no)).get()->getHistory();
	}
//	fprintf(stderr, " --- copied %d entries ---\n", (int)entries.size() );
//	fprintf(stderr," *** 7 *** cpSeqno: %d, stableCPSeqno: %d, prepareSeqno: %d, preprepareSeqno: %d\n",
//				   cpSeqno, stableCPSeqno, prepareSeqno, preprepareSeqno);
//	committedCP.get()->printInfo();
//	stableCP.get()->printInfo();	
	//finally create the view change
	boost::shared_ptr<messages::ViewChange> vc (
		new messages::ViewChange( currentView, 
						      cpSeqno,
						      stableCPSeqno,
						      prepareSeqno,
						      preprepareSeqno,
						      committedCP,
						      stableCP,
						      entries,
						      histories,
						      baseNode.get()->getMyOrderIndex() ) );

	//Verify the CP correctness
//	fprintf(stderr, " +++ Verify the CP correctness\n");
//	int temp_cp_bytes_size;
//	boost::shared_ptr<char> temp_cp_bytes = committedCP.get()->getBytes(temp_cp_bytes_size);
//	boost::shared_ptr<statemanagement:: CheckPointState> new_committedCP (
//		new statemanagement::CheckPointState(temp_cp_bytes, temp_cp_bytes_size)	);
//	new_committedCP.get()->printInfo();
	
	//Verify vc correctness
//	fprintf(stderr, " +++ Verify the vc correctness\n");
//	int new_vc_bytes_size;
//	boost::shared_ptr<char> new_vc_bytes = vc.get()->getBytes(new_vc_bytes_size);
//	boost::shared_ptr<messages::ViewChange> new_vc(  new messages::ViewChange(new_vc_bytes, new_vc_bytes_size));
//	new_vc.get()->printInfo();

//	fprintf(stderr," *** 8 *** cpSeqno: %d, cpIndex: %d\n",cpSeqno, cpIndex);							 
	baseNode.get()->authenticateOrderMacArrayMessage(vc);
	int vc_bytes_size;
	boost::shared_ptr<char> vc_bytes = vc.get()->getBytes(vc_bytes_size);
	baseNode.get()->sendToAllOrder(vc_bytes, vc_bytes_size);
	
	//create the new view information
	vInfo.reset (
		new statemanagement::ViewInfo( currentView, vcCerts, primary(currentView) == baseNode.get()->getMyOrderIndex())	);
	
	changingView = true;
	vInfo.get()->observeViewChange(vc);
	vInfo.get()->setMyViewChange(vc);
	
	//create a vcack for my vc
	
	boost::shared_ptr<BFT::messages::Digest> d (new BFT::messages::Digest(vc_bytes, vc_bytes_size));
	boost::shared_ptr<messages::ViewChangeAck> vca (
		new messages::ViewChangeAck( vc.get()->getView(),
				   vc.get()->getSendingReplica(),
				   d,
				   baseNode.get()->getMyOrderIndex() ));
	//verifying vcack correctness
	fprintf(stderr, "OrderBaseNode:process(viewChange) says : created vcack. sender %d, changingreplica %d\n",
			vca.get()->getSender(), vca.get()->getChangingReplica());
	
	baseNode.get()->authenticateOrderMacArrayMessage(vca);
	if(!amIPrimary(currentView))
	{
		int vca_bytes_size;
		boost::shared_ptr<char> vca_bytes = vca.get()->getBytes(vca_bytes_size);
		baseNode.get()->sendToOrder(vca_bytes, vca_bytes_size, primary(currentView));
		fprintf(stderr, "OrderBaseNode:process(viewChange) says : sent vcack. sender %d, recipient %d changingreplica %d\n",
				vca.get()->getSender(), primary(currentView),vca.get()->getChangingReplica());
	}
	else
	{
		fprintf(stderr, "OrderBaseNode:process(viewChange) says : adding MY vcack. sender %d, changingreplica %d\n",
				vca.get()->getSender(), vca.get()->getChangingReplica());
		vInfo.get()->addViewChangeAck(vca);
	}
	
	//look for cached new view
	if(nvCache.at( primary(currentView) ).use_count() != 0 &&
	   nvCache.at( primary(currentView) ).get()->getView() == currentView )
	{
		process( nvCache.at( primary(currentView) ) );
		nvCache.at(primary(currentView)).reset();
	}
}	


void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::ViewChange> _vc)
{


	uint32_t vc_view = _vc.get()->getView();
	uint32_t vc_sender = _vc.get()->getSender();

	fprintf(stderr,"OrderBaseNode::process(VC) says : VC - view %d from %d\n",vc_view, vc_sender);
	
	// (1)  if the view is old, or smaller than the last received from the sender, 
	// or its for the current view but we're done view changing already, then dont do anything.

	if( vc_view < currentView || 
	    vc_view < viewChanges.at(vc_sender).get()->getLastReceived() ||
	    (vc_view == currentView && !changingView) )
	{
		fprintf(stderr,"OrderBaseNode::process(ViewChange) says: discarding an old view change\n");
		return;
	}
	
	if(!baseNode.get()->validateOrderMacArrayMessage(_vc))
	{
		fprintf(stderr,"OrderBaseNode::process(viewchange) says: view change did not validate!\n");
		exit(165);
	}

	boost::shared_ptr<BFT::messages::Digest> d (
		new BFT::messages::Digest( _vc.get()->getBytes() , DIGEST_SIZE ) );
	
	boost::shared_ptr<messages::ViewChangeAck> vcack (
		new messages::ViewChangeAck ( vc_view, vc_sender, d, baseNode.get()->getMyOrderIndex() ) );
		
	baseNode.get()->authenticateOrderMacArrayMessage(vcack);
	
	if ( primary(vc_view) != baseNode.get()->getMyOrderIndex())
	{
		int vcack_bytes_size;
		boost::shared_ptr<char> vcack_bytes =  vcack.get()->getBytes(vcack_bytes_size);
		baseNode.get()->sendToOrder(vcack_bytes, vcack_bytes_size, primary(vc_view));
	}
	fprintf(stderr, "OrderBaseNode:process(viewChange) says : sent vcack. sender %d, recipient %d changingreplica %d\n",
			vcack.get()->getSender(), primary(vc_view),vcack.get()->getChangingReplica());
	// if the view change is for the current view, then add it to
	// the view info

	if( vc_view == currentView)
	{
//		fprintf(stderr, "OrderBaseNode::process(ViewChange) says:  View Change for view %d from %d is for current view\n"
//				,vc_view, vc_sender);
		if(vInfo.use_count() == 0)
		{
			fprintf(stderr, "AAAAAAAAAAARRRRRRRRRGHHHHGGHHHH!\n");
			exit(166);
		}		
		vInfo.get()->observeViewChange(_vc);
		//if I am the primary, record the ack
		if( primary(currentView) == baseNode.get()->getMyOrderIndex())
		{
			fprintf(stderr, "OrderBaseNode:process(viewChange) says : adding MY vcack. sender %d, changingreplica %d\n",
				vcack.get()->getSender(),vcack.get()->getChangingReplica());

			vInfo.get()->addViewChangeAck(vcack);
			
			if(vInfo.get()->getNewViewMessage().use_count() == 0)
			{
				boost::shared_ptr<messages::NewView> nv = vInfo.get()->generateNewViewMessage(baseNode);
				if( nv.use_count() != 0)
				{
					if( !vInfo.get()->definedView() )
					{
						fprintf(stderr,"OrderBaseNode::process(ViewChange) says: primary better have a \
								defined view if it creates a newview message.\n");
						exit(166);
					}
					else
					{
						actOnDefinedView();
					}
				}
			}
		}
		else
		{
			if(vInfo.get()->definedView() )
			{
				actOnDefinedView();
				return;
			}
		}
	}
	else
	{
		//View Change from the future
//		fprintf(stderr, "OrderBaseNode::process(ViewChange) says:  View Change for view %d from %d is for future view\n"
//				,vc_view, vc_sender);

		viewChanges.at(vc_sender).get()->observeViewChange(_vc, primary(vc_view) == baseNode->getMyOrderIndex() );
		
		//record the ack if I am the primary
		if (primary(vc_view) == baseNode.get()->getMyOrderIndex())
		{
//			fprintf(stderr,"\t***** %d *****\n", 
//				viewChanges.at(vc_sender).get()->getCertificate(vc_view).get()->getViewChange().get()->getSender() );

			fprintf(stderr, "OrderBaseNode:process(viewChange) says : adding MY vcack. sender %d, changingreplica %d\n",
				vcack.get()->getSender(), vcack.get()->getChangingReplica());
			viewChanges.at(vc_sender).get()->addViewChangeAck(vcack);
		}
		
		//if f+1 replicas are advocating a view larger than currentView, go
		// to the smallest of those views
		
		uint32_t small = currentView;
		uint32_t temp_smallest;
		int count = 0;
		for(unsigned int i=0; i< viewChanges.size(); i++)
		{
			temp_smallest = viewChanges.at(i).get()->getSmallestView();
			
			if( temp_smallest > currentView )
			{
				count++;
			}
			
			if ( ( temp_smallest < small || small == currentView) &&
			     temp_smallest != currentView )
			{
				small = temp_smallest;
			}	      
		}
		
		if (count >= BFT::order::Parameters::smallOrderQuorumSize()  ||
		    viewChanges.at( primary(currentView) ).get()->getSmallestView() >= small  )
		{
			fprintf(stderr,"OrderBaseNode::process(ViewChange) says: starting a view change b/c it came in the mail\n" );
			startViewChange(small);
		} 
	}
}			


//The purpose of this function is to ensure that the replica has
//all of the state specified by the new view message.
void OrderBaseNode::actOnDefinedView()
{
//	fprintf(stderr, "OrderBaseNode::actOnDefinedView says: acting on defined view!\n");
	if (!vInfo.get()->definedView())
	{
		fprintf(stderr,"OrderBaseNode::actondefinedview says: view is not defined\n");
		exit(170);
	}

	bool allInformation = true;
	// check that the committed checkpoint is present
	int ind = baseIndex;
	int indplusone = (baseIndex+1) % stateSnapshots.size();
	//System.out.println("commitcp DIGEST: "+vInfo.getCommittedCPDigest());
	vInfo.get()->addMyCP(stateSnapshots.at(ind));
	vInfo.get()->addMyCP(stateSnapshots.at(indplusone));

	// check that all specified pieces are present
	std::vector<bool> missing = vInfo.get()->getMissingRequests();
	uint32_t base = vInfo.get()->getBaseSequenceNumber();

	int missingop = 0;
	// we can only build the cp if we dont need the cp to be
	// stable
	bool canbuildcp = ( (vInfo.get()->getNextSeqNo()- vInfo.get()->getBaseSequenceNumber() ) < 
			    (uint32_t)(BFT::order::Parameters::maxPeriods * BFT::order::Parameters::checkPointInterval) ) 
	    		  && (vInfo.get()->getStableCPDigest().use_count() == 0);
//	fprintf(stderr,"OrderBaseNode::actondefinedview says: initial can build: %d\n",canbuildcp?1:0);

	// for all mising requests, see if we have it locally, if so
	// load it up
	for (unsigned int i = 0; i < missing.size(); i++)
	{
		//fprintf(stderr,"OrderBaseNode::actondefinedview says: is req %d is missing? %d\n",base+i, missing.at(i)?1:0);
		if (missing.at(i))
		{
			boost::shared_ptr<statemanagement::Certificate> cert = 
				certificates.at(stateIndex(base+i)).at(certificateIndex(base+i));
			//System.out.println("checking on: "+(base+i));
			if (!vInfo.get()->setCertificate(cert, base+i))
			{
				missingop++;
				fprintf(stderr,"OrderBaseNode::actondefinedview says: failed to add the entry at :%d\n",(base+i));
				canbuildcp = false;
			}
			else
			{
				fprintf(stderr,"OrderBaseNode::actondefinedview says: added the entry at :%d\n",(base+i));
				missing[i] = false;
			}
		}
		
		allInformation = allInformation && !missing.at(i);
	}

	// check to see if we need to fetch the committed cp
	boost::shared_ptr<statemanagement::CheckPointState> committed = vInfo.get()->getCommittedCP();
	if (committed.use_count() != 0)
	{
		makeCPStable(committed);
	}

	if ( committed.use_count() == 0 || 
	     ( !committed.get()->isStable() && !committed.get()->isMarkingStable()) )
	{
		fprintf(stderr,"OrderBaseNode::actondefinedview says: sending missing cp\n");
		boost::shared_ptr<messages::MissingCP> mcp 
			( new messages::MissingCP(vInfo.get()->getBaseSequenceNumber(),
					  vInfo.get()->getCommittedCPDigest(),
					  baseNode.get()->getMyOrderIndex() ) );
		baseNode.get()->authenticateOrderMacArrayMessage(mcp);

		int mcp_bytes_size;
		boost::shared_ptr<char> mcp_bytes = mcp.get()->getBytes(mcp_bytes_size);

		if (amIPrimary(currentView))
		{
			baseNode.get()->sendToAllOrder(mcp_bytes, mcp_bytes_size);
		}
		else
		{
			baseNode.get()->sendToOrder(mcp_bytes, mcp_bytes_size, primary(currentView));
		}
		
		allInformation = false;
		fprintf(stderr,"OrderBaseNode::actondefinedview says: cant build it up b/c the committed cp is not stable\n");
		canbuildcp = false;
	}

	if (committed.use_count() != 0 && !committed.get()->isCommitted() && committed.get()->isStable())
	{
		committed.get()->commit();
	}

	// check to see if we need to fetch the stable cp
	boost::shared_ptr<statemanagement::CheckPointState> stable = vInfo.get()->getStableCP();
	if ( (vInfo.get()->getStableCPDigest().use_count() != 0)  &&  (stable.use_count() != 0))
	{
		makeCPStable(stable);
	}
	
	if (vInfo.get()->getStableCP().use_count() == 0)
	{
		// if we have the committed checkpoint and sufficient entries,
		// build up the stable cp
		//System.out.println("can build up the cp?"+canbuildcp);
		if (canbuildcp)
		{
			//System.out.println(vInfo.getCommittedCP());
			boost::shared_ptr<statemanagement::CheckPointState> stable (
				new statemanagement::CheckPointState(committed) );
				
			std::vector<boost::shared_ptr<statemanagement::Certificate> > entries = vInfo.get()->getEntries(0);
			for (unsigned int i = 0; i < entries.size(); i++)
			{
				if (entries.at(i).get()->getCertEntry().use_count() == 0)
				{
					fprintf(stderr,"OrderBaseNode::actondefinedview says : Entry %d sould not be null!",i);
		    		}
		    		
		    		stable.get()->addNextBatch(entries.at(i).get()->getCertEntry(), 
		    					   stable.get()->getBaseSequenceNumber()+i);
			}
		
			vInfo.get()->addMyCP(stable);
			fprintf(stderr,"OrderBaseNode::actondefinedview says: built and added stableCP: %d\n",
				stable.get()->getBaseSequenceNumber());
			
			//TODO do not understand why we do that
			boost::shared_ptr<BFT::messages::RequestCP> rcp( 
				new BFT::messages::RequestCP(stable.get()->getCurrentSequenceNumber(),
					      		baseNode.get()->getMyOrderIndex()) );
					      		
			baseNode.get()->authenticateExecMacArrayMessage(rcp);
			int rcp_bytes_size;
			boost::shared_ptr<char> rcp_bytes = rcp.get()->getBytes(rcp_bytes_size);
			baseNode.get()->sendToAllExec(rcp_bytes, rcp_bytes_size);
			
			if (vInfo.get()->getStableCP().use_count() == 0)
			{
		    		fprintf(stderr,"OrderBaseNode::actondefinedview says : WTF?\n");
		    		exit(171);
		    	}
		}
		else
		{
			if (vInfo.get()->getStableCPDigest().use_count() != 0)
			{
				boost::shared_ptr<messages::MissingCP> mcp(
					new messages::MissingCP(vInfo.get()->getBaseSequenceNumber()
									+ BFT::order::Parameters::checkPointInterval,
					      			vInfo.get()->getStableCPDigest(),
					      			baseNode.get()->getMyOrderIndex() ) );
					      			
				baseNode.get()->authenticateOrderMacArrayMessage(mcp);
				fprintf(stderr,"OrderBaseNode::actondefinedview says: sending missing cp for %d\n",
					mcp.get()->getSeqNo());
				int mcp_bytes_size;
				boost::shared_ptr<char> mcp_bytes = mcp.get()->getBytes(mcp_bytes_size);
				if (amIPrimary(currentView))
				{
		    			baseNode.get()->sendToAllOrder(mcp_bytes, mcp_bytes_size);
		    		}
				else
				{
		    			baseNode.get()->sendToOrder(mcp_bytes, mcp_bytes_size, primary(currentView));
				}
				
				allInformation = false;
			}
		}
	}

//	fprintf(stderr,"OrderBaseNode::actondefinedview says: checking for missing ops...\n");

	// fetch the requisit operations
	if (missingop > 0)
	{
		std::vector<uint32_t> ops;
		//ops.resize(missingops);
		
		//TODO that seems to be wrong
//		for (unsigned int i = 0; i < ops.size(); i++)
//		{
//			for(unsigned int j = 0; j < missing.size(); j++)
//			{
//		    		if (missing.at(j))
//		    		{
//					ops.at(i) = j+base;
//					break;
//		    		}
//		    	}
//		}
		
		for(unsigned int i = 0; i< missing.size(); i++)
		{
			if(missing.at(i))
			{
				ops.push_back(i+base);
			}
		}
		
		//just a consistency check
		if (missingop != (int)ops.size())
		{
			fprintf(stderr,"OrderBaseNode::actondefinedview says: problems in determining which ops are missing\n");
			exit (171);
		}
		
		fprintf(stderr,"OrderBaseNode::actOnDefinedView says: %d ops are missing\n",(int)ops.size());

		boost::shared_ptr<messages::MissingOps> mo ( 
			new messages::MissingOps(currentView, ops, baseNode.get()->getMyOrderIndex()) );
			
		baseNode.get()->authenticateOrderMacArrayMessage(mo);
		
		int mo_bytes_size;
		boost::shared_ptr<char> mo_bytes = mo.get()->getBytes(mo_bytes_size);
		
		if (amIPrimary(currentView))
		{
			baseNode.get()->sendToAllOrder(mo_bytes, mo_bytes_size);
		}
		else
		{
			baseNode.get()->sendToOrder(mo_bytes, mo_bytes_size, primary(currentView));
		}
	}

	// if we have everything
	if (allInformation
	    && vInfo.get()->getCommittedCP().get()->isCommitted() 
	    && vInfo.get()->getStableCP().use_count() != 0
	    && (vInfo.get()->getStableCPDigest().use_count() == 0 
		|| vInfo.get()->getStableCP().get()->isStable()))
	{
		fprintf(stderr,"OrderBaseNode::actondefinedview says: view is complete\n");
	    	actOnCompleteView();
	}
	else
	{
		fprintf(stderr,"OrderBaseNode::actondefinedview says: view is *NOT* complete\n"); 
		if (!allInformation)
		{
			fprintf(stderr, "OrderBaseNode::actondefinedview says: missing some bit of information\n");
		}
		
		if (! vInfo.get()->getCommittedCP().get()->isCommitted() )
		{
			fprintf(stderr, "OrderBaseNode::actondefinedview says: committed CP is not stable\n");
		}
		
		if(vInfo.get()->getStableCPDigest().use_count() != 0)
		{
			fprintf(stderr,"OrderBaseNode::actondefinedview says: there is no stable CP digest\n");
		}
		
		if(!vInfo.get()->getStableCP().get()->isStable())
		{
			fprintf(stderr, "OrderBaseNode::actondefinedview says: stableCPDigest is not stable\n");
		}	
	}
}

//we have all the information for the new view
void OrderBaseNode::actOnCompleteView()
{
//	fprintf(stderr, "OrderBaseNode::actOnCompleteView not implemented yet!!!\n");
	if (!vInfo.get()->isComplete())
	{
		fprintf(stderr,"OrderBaseNode::actoncompleteview says: view is not complete, and I am confused\n");
		exit(172);
	}

	// if i've already sent a confirmation, no need to do it again
	if (vInfo.get()->confirmedBy(baseNode.get()->getMyOrderIndex()))
	{
//		fprintf(stderr,"OrderBAseNode::actoncompleteview says: I've already confirmed\n");
		return;
	}

	if (amIPrimary(currentView))
	{
		boost::shared_ptr<messages::NewView> nv = vInfo.get()->getNewViewMessage();
		if (nv.use_count() == 0)
		{
			fprintf(stderr,"OrderBaseNode::actoncompleteview says: nv does not exist, and I am confused\n");
			exit(173);
		}	
		baseNode.get()->authenticateOrderMacArrayMessage(nv);
		int nv_bytes_size;
		boost::shared_ptr<char> nv_bytes = nv.get()->getBytes(nv_bytes_size);
		baseNode.get()->sendToAllOrder(nv_bytes, nv_bytes_size);
		fprintf(stderr,"OrderBaseNode::actOnCompleteView says: sent NewView: view: %d, chosenvcs %d %d %d %d\n"
			, nv.get()->getView(), nv.get()->getChosenVCs().at(0).use_count()?1:0
			, nv.get()->getChosenVCs().at(1).use_count()?1:0 , nv.get()->getChosenVCs().at(2).use_count()?1:0
			, nv.get()->getChosenVCs().at(3).use_count()?1:0);
	}


	boost::shared_ptr<messages::ConfirmView> cv ( 
		new messages::ConfirmView(currentView,
					 vInfo.get()->getNextSeqNo(),
					 vInfo.get()->getLastHistory(),
					 baseNode.get()->getMyOrderIndex()) );
	baseNode.get()->authenticateOrderMacArrayMessage(cv);
	vInfo.get()->addConfirmation(cv);
	int cv_bytes_size;
	boost::shared_ptr<char> cv_bytes = cv.get()->getBytes(cv_bytes_size);
	baseNode.get()->sendToAllOrder(cv_bytes, cv_bytes_size);
	fprintf(stderr,"OrderBaseNode::actOnCompleteView says: sent confirm view: view: %d, nextseqno: %d\n"
		, cv.get()->getViewNo(), cv.get()->getSeqNo());
	vInfo.get()->flushConfirmedCache();

	// copy all of the state over from our caches into the 'live'
	// realm and mark all requests as prepared.
	
	if (cachedCPT.use_count() != 0)	
	{
		boost::shared_ptr<statemanagement::CheckPointState> cps = vInfo.get()->getStableCP();
		if ( cps.use_count() != 0 &&
		     cps.get()->getCurrentSequenceNumber() == cachedCPT.get()->getSequenceNumber())
		{
			int token_size;
			boost::shared_ptr<char> token_bytes;
			token_bytes = cachedCPT.get()->getToken(token_size);
			cps.get()->addExecCPToken(token_bytes, token_size, cachedCPT.get()->getSequenceNumber());
			makeCPStable(cps);
		}
		else
		{
			cps = vInfo.get()->getCommittedCP();
			if (cps.use_count() != 0 &&
		    	cps.get()->getCurrentSequenceNumber() == cachedCPT.get()->getSequenceNumber())
		    	{
				int token_size;
				boost::shared_ptr<char> token_bytes;
				token_bytes = cachedCPT.get()->getToken(token_size);
				cps.get()->addExecCPToken(token_bytes, token_size, cachedCPT.get()->getSequenceNumber());
				makeCPStable(cps);
			}
			else
			{
				fprintf(stderr, "OrderBaseNode::actoncptoken something wrong there...\n");
				exit(184);
			}
		}
		
		cachedCPT.reset();

	}
	
	int baseplusone = (baseIndex+1) % stateSnapshots.size();
	stateSnapshots.at(baseIndex) = vInfo.get()->getCommittedCP();
	if (!stateSnapshots.at(baseIndex).get()->isStable())
	{
		fprintf(stderr,"OrderBaseNode::actoncompleteview says: should be stable already!\n");
		exit(173);
	}
	stateSnapshots.at(baseIndex).get()->commit();
	certificates.at(baseIndex) = vInfo.get()->getEntries(0);
	stateSnapshots.at(baseplusone) = vInfo.get()->getStableCP();
	certificates.at(baseplusone) = vInfo.get()->getEntries(1);
	if (vInfo.get()->getStableCP().use_count() == 0)
	{
		fprintf(stderr,"OrderBaseNode::actoncompleteview says: vInfo has no stable CP, and i am confused\n");
		exit(174);
	}

	workingState = vInfo.get()->getWorkingState();
	fprintf(stderr,"base seqno %d, current seqno %d\n",getBaseSequenceNumber(),workingState.get()->getCurrentSequenceNumber());
	//checking certificates

	// mark everything as prepared
	for (uint32_t i = getBaseSequenceNumber(); i < getCurrentSequenceNumber(); i++)
	{
		if (!certificates.at(stateIndex(i)).at(certificateIndex(i)).get()->isPrepared())
		{
			certificates.at(stateIndex(i)).at(certificateIndex(i)).get()->forcePrepared();
//			//TODO huh?
//			fprintf(stderr, "marking seqno %d, stateIndex %d, certificateIndex %d as prepared\n",
//				i, stateIndex(i), certificateIndex(i));
//			//System.out.println("\tNeed to send out prepare equivalent NB here");
		}
	}
	
	maxPrepared = getCurrentSequenceNumber() - 1;

	if (vInfo.get()->isConfirmed())
	{
		actOnConfirmedView();
	}
//	else
//	{
//		fprintf(stderr,"OrderBaseNode::actondefinedview says: view is *NOT* confirmed\n");
//	}


}

void OrderBaseNode::process(boost::shared_ptr<messages::ConfirmView> _cv)
{
	if (_cv.get()->getViewNo() < currentView)
	{
		fprintf(stderr,"OrderBaseNode::process(confirmview) says: ignoring old confirmview\n");
		return;
	}

	if (vInfo.get()->isConfirmed())
	{
		fprintf(stderr,"OrderBaseNode::process(confirmview) says: view already confirmed\n");
		return;
	}

	if (!baseNode.get()->validateOrderMacArrayMessage(_cv))
	{
		fprintf(stderr,"OrderBaseNode::process(confirmview) says: confirmview did not validate\n");
	    	exit(176);
	}

	if (_cv.get()->getViewNo() != currentView)
	{
		fprintf(stderr,"OrderBaseNode::process(confirmview) says: out of order confirm view!!!!!\n");
	    	viewChanges.at((int)_cv.get()->getSender()).get()->addConfirmView(_cv,
								    	amIPrimary(_cv.get()->getViewNo()) );
		//TODO						    	
		// and i think thats it?  though it might entail doing the
		// 'view change' check as well -- i.e. process(view
		// change) looks to see if this message will trigger
		// another future vc
		return;
	}
	fprintf(stderr,"OrderBaseNode::process(confirmview) says: received CV for view %d, seqno %d, sender %d\n"
		, _cv.get()->getViewNo(), _cv.get()->getSeqNo(), _cv.get()->getSender());
	vInfo.get()->addConfirmation(_cv);
	if (vInfo.get()->definedView() && vInfo.get()->isComplete() && vInfo.get()->isConfirmed())
	{
		fprintf(stderr,"OrderBaseNode::process(confirmview) says: acting on confirmed view!\n");
		actOnConfirmedView();
	}
	else
	{
		fprintf(stderr,"OrderBaseNode::process(confirmview) says: *NOT* acting on confirmed view...\n");
		
		if(!vInfo.get()->definedView())
		{
			fprintf(stderr,"OrderBaseNode::process(confirmview) says: view not defined.\n");
		}
		
		if(!vInfo.get()->isComplete())
		{
			fprintf(stderr,"OrderBaseNode::process(confirmview) says: view not complete.\n");
		}

		if(!vInfo.get()->isConfirmed())
		{
			fprintf(stderr,"OrderBaseNode::process(confirmview) says: view not confirmed.\n");
		}

	}
}

void OrderBaseNode::actOnConfirmedView()
{
	if (!vInfo.get()->isConfirmed())
	{
		fprintf(stderr,"OrderBaseNode::actOnConfirmedView says: view should be confirmed if you're getting here\n");
		exit(179);
	}
	
	if (stateSnapshots.at(baseIndex).get()->getStableDigest().use_count() == 0)
	{
		fprintf(stderr,"OrderBaseNode::actOnConfirmedView says: cannot confirm view with unstable base!\n");
		exit(180);
	}

	int baseplusone = (baseIndex + 1) % stateSnapshots.size();

	int lastIndex = BFT::order::Parameters::checkPointInterval -1;
	
	if (stateSnapshots.at(baseplusone).get()->getStableDigest().use_count() == 0 
	    && certificates.at(baseplusone).at(lastIndex).get()->getHistory().use_count() != 0)
	{
		fprintf(stderr,"OrderBaseNode::actOnConfirmedView says: cannot confirm view. second checkpoint must be stable, but it is not\n");
		exit(181);
	}

	if (workingState.get()->getCurrentSequenceNumber() != vInfo.get()->getNextSeqNo()
	    || !workingState.get()->getHistory().get()->equals(vInfo.get()->getLastHistory()))
	{
		fprintf(stderr,"OrderBaseNode::actOnConfirmedView says: cannot confirm view. working state does not match the current sequence of requests\n");
		exit(181);
	}

	if (!amIPrimary(currentView))
	{
	    	boost::shared_ptr<messages::StartView> sv ( 
	    		new messages::StartView(currentView, 
					getCurrentSequenceNumber(),
					workingState.get()->getHistory(), 
					baseNode.get()->getMyOrderIndex()));
					
	    	baseNode.get()->authenticateOrderMacMessage(sv, primary(currentView));
	    	int sv_bytes_size;
	    	boost::shared_ptr<char> sv_bytes = sv.get()->getBytes(sv_bytes_size);
	    	baseNode.get()->sendToOrder(sv_bytes, sv_bytes_size, primary(currentView));
		fprintf(stderr,"OrderBaseNode::actOnConfirmedView says: sent SV for view %d, seqno %d\n"
			,sv.get()->getView(), sv.get()->getSeqNo());
	}
	// mark everything as committed
	for (uint32_t i = getBaseSequenceNumber(); i < getCurrentSequenceNumber(); i++)
	{
		certificates.at(stateIndex(i)).at(certificateIndex(i)).get()->forceCommitted();
	}
	
	changingView = false;
	fprintf(stderr,"OrderBaseNode::actOnConfirmedView says: not changing view anymore\n");
	
}

void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::StartView> _sv)
{
	// if its not for this view or im not the primary, drop the
	// message
	if (_sv.get()->getView() != currentView || !amIPrimary(currentView))
	{
		fprintf(stderr,"OrderBaseNode::process(startview) says: dropping useless startview\n");
		return;
	}
	
	// the sender is just too far behind to be helped!
	if (_sv.get()->getSeqNo() < getBaseSequenceNumber())
	{
		fprintf(stderr,"OrderBaseNode::process(startview) says: dropping startview, sender too far behind\n");
		return;
	}
	
	fprintf(stderr,"OrderBaseNode::process(startview) says: received SV for view %d, seqno %d - rate limiting goes here\n"
		,_sv.get()->getView(), _sv.get()->getSeqNo());

	if (!baseNode.get()->validateOrderMacMessage(_sv) )
	{
		fprintf(stderr,"OrderBaseNode::process(startview) says: startview, did not validate\n");
		return;
	}

	uint32_t ind = vInfo.get()->getNextSeqNo();
	while (ind < getCurrentSequenceNumber())
	{
		boost::shared_ptr<messages::PrePrepare> pp = 
			certificates.at(stateIndex(ind)).at(certificateIndex(ind)).get()->getPrePrepare();
		if (pp.use_count() == 0)
		{
			fprintf(stderr,"OrderBaseNode::process(startview) says: pp should not be null in this view!\n");
			exit(182);
		}
		
		int pp_bytes_size;
		boost::shared_ptr<char> pp_bytes = pp.get()->getBytes(pp_bytes_size);
		baseNode.get()->sendToOrder(pp_bytes, pp_bytes_size, _sv.get()->getSendingReplica());
		ind++;
	}
}

void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::NewView> _nv)
{
	if (_nv.get()->getView() < currentView)
	{
		fprintf(stderr,"OrderBaseNode::process(NewView) says : ignoring old newview\n");//return;
		return;
	}

	if (!baseNode.get()->validateOrderMacArrayMessage(_nv))
	{
		fprintf(stderr,"NewView message did not validate\n");//return;
		exit(174);
	}
	
	if (_nv.get()->getView() > currentView)
	{
		int index = (int)(_nv.get()->getSendingReplica());
		if ( nvCache.at(index).use_count() == 0 ||
		     nvCache.at(index).get()->getView() < _nv.get()->getView())
		{
			fprintf(stderr,"OrderBaseNode::process(NewView) says : caching future newview. currentView %d nv.view %d\n"
					, currentView, _nv.get()->getView());//return;
			nvCache.at(index) = _nv;
		}
		return;
	}

	if (!changingView)
	{
		fprintf(stderr,"OrderBaseNode::process(NewView) says : ignoring newview, b/c we are not changing view\n");//return;
		return;  // we're no longer changing view!
	}

	if (vInfo.get()->getNewViewMessage().use_count() == 0)
	{
		fprintf(stderr,"OrderBaseNode::process(NewView) says : Adding newview - view: %d, sender %d, chosenvcs %d %d %d %d\n"
			, _nv.get()->getView(), _nv.get()->getView(), _nv.get()->getChosenVCs().at(0).use_count()!=0?1:0
			, _nv.get()->getChosenVCs().at(1).use_count()!=0?1:0 , _nv.get()->getChosenVCs().at(2).use_count()!=0?1:0
			, _nv.get()->getChosenVCs().at(3).use_count()!=0?1:0);
		vInfo.get()->addNewView(_nv);
			
			
		// if the view is defined, then act on it
		if (vInfo.get()->definedView())
		{
			actOnDefinedView();
		}
		else
		{
			fprintf(stderr,"OrderBaseNode::process(NewView) says : view: %d bot defined yet. check for missing VCs\n",
				_nv.get()->getView());
			std::vector<bool> mv = vInfo.get()->getMissingViewChanges();
			for(unsigned int i = 0; i < mv.size(); i++)
			{
				if (mv.at(i))
				{
					boost::shared_ptr<messages::MissingViewChange> mvc (
						new messages::MissingViewChange(currentView, i, baseNode.get()->getMyOrderIndex()) );
						
					baseNode.get()->authenticateOrderMacMessage(mvc,primary(currentView));
					
					int mvc_bytes_size;
					boost::shared_ptr<char> mvc_bytes = mvc.get()->getBytes(mvc_bytes_size);
					
					baseNode.get()->sendToOrder(mvc_bytes, mvc_bytes_size, primary(currentView));
					fprintf(stderr,"OrderBaseNode::processnewview says : fetching vc for : %d\n",i);
				}
			}
		}
	}
}

void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::ViewChangeAck> _vca)
{
	uint32_t vca_view = _vca.get()->getViewNo();
//	uint32_t vca_sender = _vca.get()->getSender();

	fprintf(stderr,"OrderBaseNode::process(viewchangeack) says: processing vca from %d for view %d, changingreplica %d\n"
		,_vca.get()->getSender(), vca_view, _vca.get()->getChangingReplica());
	
	if(vca_view < currentView)
	{
		return;
	}
	else
	{
		if( primary(vca_view) != baseNode.get()->getMyOrderIndex() )
		{
			fprintf(stderr,"OrderBaseNode::process(viewchangeack) says: not the primary for vca view!\n");
			exit(167);
		}
		else
		{
			if(!baseNode.get()->validateOrderMacArrayMessage(_vca))
			{
				fprintf(stderr,"OrderBaseNode::process(viewchangeack) says: vca does not validate!\n");
				exit(168);
			
			}
			else
			{
				if( vca_view > currentView)
				{
					//cachine a vc ack for a future view
					viewChanges.at(_vca.get()->getChangingReplica()).get()->addViewChangeAck(_vca);
				}
				else
				{
					// vca_view !< currentView && vca_view !> currentView, no need to check
					// whether vca_view == currentView
					vInfo.get()->addViewChangeAck(_vca);
					if ( vInfo.get()->getNewViewMessage().use_count() == 0 )
					{
						boost::shared_ptr<messages::NewView> nv = vInfo.get()->generateNewViewMessage(baseNode);
						if (nv.use_count()!=0)
						{
							actOnDefinedView();
						}
						else
						{
//							fprintf(stderr,"OrderBaseNode::process(viewchangeack) says: nv.usecount still == 0\n");
						}
					}
				}
			}
		}
	}		
}

//TODO
//void process(boost::shared_ptr<BFT::order::messages::MissingViewChange> _mvc);
//void process(boost::shared_ptr<BFT::order::messages::RelayViewChange> _rvc);
//void process(boost::shared_ptr<BFT::order::messages::MissingOps> _mo);
//void process(boost::shared_ptr<BFT::order::messages::RelayOps> _ro);
//void process(boost::shared_ptr<BFT::order::messages::StartView> _sv);

void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::MissingCP> _mcp)
{
	fprintf(stderr, "NIY 7\n");
}

void OrderBaseNode::process(boost::shared_ptr<BFT::order::messages::RelayCP> _rcp)
{
	fprintf(stderr, "NIY 8\n");
}


//TODO
/*
A clean request core is processed following the following steps:
(1) If the request core has been ordered, then discard it
(2) If I am not the primary and the requestcore has not appeared in a preprepare, then forward to the primary
(3) If I am the primary and the request core has not yet appeared in a preprepare then add the request core to the request queue and attempt to form a new batch. if a new batch can be formed, then send the resulting pre prepare.
(4) Initial setup step: use the first request as a hint to go fetch the intiail state. This is a short term fix that NEEDS to be addressed 
*/
void OrderBaseNode::processClean(boost::shared_ptr<std::vector<boost::shared_ptr<BFT::messages::RequestCore> > > _rcs)
{
	//if I am not the primary, forward to the primary
	if(!amIPrimary(currentView))
	{
//		fprintf(stderr, "dropping a request core because i am not the primary (TODO)\n");
		return;
		//TODO do not want to deal with forwared requests right now
	}
	
	createNextBatch(_rcs);
}

void OrderBaseNode::createNextBatch(boost::shared_ptr<std::vector<boost::shared_ptr<BFT::messages::RequestCore> > > _rcs)
{
	if (!amIPrimary(currentView))
	{
		fprintf(stderr, "not the primary, should not create a batch\n");
		exit(122);
	}
	
	uint32_t seqno = getCurrentSequenceNumber();
//	fprintf(stderr, "inside createnextbatch\n");
	//check if the base is stable
	if (!stateSnapshots[baseIndex].get()->isStable())
	{
		if(seqno == 0)
		{
			fprintf(stderr, "orderbasenode::createnextbatch says : should not make it here\n");
			exit(123);
		}
		else
		{
			fprintf(stderr, "orderbasenode::createnextbatch says : base should be stable\n");
			exit(124);
		}
	}
	
	//first seqno in a checkpoint
	if(seqno % BFT::order::Parameters::checkPointInterval == 0 &&
	   seqno > 0 &&
	   !certificates.at(stateIndex(seqno-1)).at(certificateIndex(seqno-1)).get()->isCommitted() )
	{
			fprintf(stderr, "orderbasenode::createnextbatch says : should not make it here\n");
			exit(125);		
	}
	
	//are there requests?
	if(_rcs.get()->size() == 0)
	{
		fprintf(stderr, "orderbasenode::createnextbatch says : empty batch?\n");
		exit(125);			
	}
	
	bool cp = (seqno + 1)%BFT::order::Parameters::checkPointInterval == 0;
	
	boost::shared_ptr<BFT::messages::Digest> d = stateSnapshots.at(baseIndex).get()->getStableDigest();
	int baseplusone = (baseIndex+1)%BFT::order::Parameters::maxPeriods;
	int baseminusone = (baseIndex + stateSnapshots.size() - 1) % stateSnapshots.size();
	if (currentIndex == baseminusone && cp)
	{
		//if the requisit cp is not stable, give up
		if(! stateSnapshots[baseplusone].get()->isStable())
		{
			fprintf(stderr, "orderbasenode::createnextbatch says : state not stable\n");
			exit(126);	
		}
		else
		{
			d = stateSnapshots.at(baseplusone).get()->getStableDigest();
		}
	}
	
	boost::shared_ptr<messages::RequestBatch> rb ( new BFT::order::messages::RequestBatch(_rcs));
	uint32_t time = millisecondTime();
	uint32_t lasttime = workingState.get()->getCurrentTime();
	if (time > lasttime)
		lasttime = time;
	else
		lasttime++;  //TODO: wall time or logical time?
	
	boost::shared_ptr<BFT::messages::NonDeterminism> nondet ( 
		new BFT::messages::NonDeterminism(lasttime, nonDetcounter++));
	boost::shared_ptr<messages::PrePrepare> pp (
		new messages::PrePrepare(currentView,
					 seqno,
					 workingState.get()->getHistory(),
					 rb,
					 nondet,
					 d,
					 baseNode.get()->getMyOrderIndex()) );
	
//	fprintf(stderr,"lasttime %ld, current time %ld, time from nondet %ld, time from pp %ld\n"
//			, lasttime, millisecondTime(), nondet.get()->getTime(), pp.get()->getNonDeterminism().get()->getTime());
					 
	//verify preprepare correctness
//	pp.get()->printInfo();
//	int pp_bytes_size;
//	boost::shared_ptr<char> pp_bytes = pp.get()->getBytes(pp_bytes_size);
//	boost::shared_ptr<messages::PrePrepare> new_pp (new messages::PrePrepare(pp_bytes, pp_bytes_size));
//	new_pp.get()->printInfo();
//	fprintf(stderr,"time from pp %ld, time from new_pp: %ld\n"
//		,pp.get()->getNonDeterminism().get()->getTime(),new_pp.get()->getNonDeterminism().get()->getTime());
					 		 
	requestsOrdered += _rcs.get()->size();
	float batchsize = (float)requestsOrdered/(seqno + 1);
	if (getCurrentSequenceNumber() % 5000 == 0)
	{
		deltaOrdered = requestsOrdered - previousOrdered;
		nowTime = millisecondTime();
		deltaTime = nowTime - previousTime;
		speed = 1000*(float)deltaOrdered/deltaTime; 
		fprintf(stdout, "avg batchsize: %f, ordered %d requests at %f req/s\n",batchsize,requestsOrdered,speed);
		previousOrdered = requestsOrdered;
		previousTime = nowTime;
	}
	baseNode.get()->authenticateOrderMacArrayMessage(pp);

	int state = stateIndex(seqno);
	int cert = certificateIndex(seqno);
	//add the preprepare to the certificate
	if(!certificates.at(state).at(cert).get()->addMyPrePrepare(pp) )
	{
		fprintf(stderr,"orderbasenode::createnextbatch says : WTF!\n");
		exit(127);
	}
	
	//send the preprepare
//	uint32_t before = millisecondTime();
	int ppsize;
	boost::shared_ptr<char> ppbytes = pp.get()->getBytes(ppsize);
//	fprintf(stderr,"orderbasenode::sending preprepare %d to all the other orders\n",pp.get()->getSeqNo());
	baseNode.get()->sendToAllOrder(ppbytes, ppsize);
//	uint32_t delta = millisecondTime() - before;
//	fprintf(stderr, "delta: %d",delta);
	
	//see if a prepare is appropriate
	boost::shared_ptr<statemanagement::Certificate> c;
	if (maxPrepared >= (int32_t)getBaseSequenceNumber())
	{
		c = certificates.at(stateIndex(maxPrepared)).at(certificateIndex(maxPrepared));
		
	}
	else
	{
		if(maxPrepared != -1)
		{
			fprintf(stderr, "orderbasenode::createnextbatch says : bad maxprepared value%d\n",maxPrepared);
			exit(128);
		}
	}
	
	//TODO more aggressive strategy for sending prepares?
	
	//create the next batch
	/*
	boost::shared_ptr<BFT::messages::CommandBatch> cb = certificates.at(state).at(cert).get()->getCommandBatch();
	boost::shared_ptr<BFT::messages::NonDeterminism> non = certificates.at(state).at(cert).get()->getNonDeterminism();
	boost::shared_ptr<BFT::messages::HistoryDigest> hist = certificates.at(state).at(cert).get()->getHistory();
	*/
	
	boost::shared_ptr<BFT::messages::NextBatch> nb = certificates.at(state).at(cert).get()
		->getNextBatch(baseNode, currentView );
	
	baseNode.get()->authenticateExecMacArrayMessage(nb);
	//add the next batch to the agreement state
	workingState.get()->addNextBatch(nb);
	
	//if it is not a checkpoint interval, rely on speculation
	if (!cp)
	{
//		fprintf(stderr,"orderbasenode::createnextbatch says : sending nb %d speculatively\n",nb.get()->getSeqNo());
		writeAndSend(nb);
	}
	
	//if the seqno is prepared, act on the prepared 
	if (certificates.at(state).at(cert).get()->isPrepared())
	{
		fprintf(stderr,"orderbasenode::createnextbatch says : before acting on prepared ************\n");
		actOnPrepared(pp.get()->getSeqNo()); 
	}
}

void OrderBaseNode::processClean(boost::shared_ptr<messages::PrePrepare> _pp)
{

	if(_pp.get()->getView() != currentView)
	{
//		fprintf(stderr,"orderbasenode::processClean(preprepare) says : dropping preprepare in a different view\n");
		return;
	}
		
	if(_pp.get()->getSeqNo() != getCurrentSequenceNumber()) 
	{
		fprintf(stderr,"orderbasenode::processClean(preprepare) says : dropping preprepare with different seqno\n");
		return;
	}
	//in the following code we assume seqno == getCurrentSequenceNumber == _pp.get()->getSeqNo!!!
	uint32_t seqno = getCurrentSequenceNumber();
	
//	fprintf(stderr, "@ received PrePrepare for seqno %d\n",seqno);

	bool cp = (seqno+1) % Parameters::checkPointInterval == 0;
	boost::shared_ptr<statemanagement::Certificate> cert = certificates.at(stateIndex(seqno)).at(certificateIndex(seqno));
	
	//0 add the preprepare
	if ( !cert.get()->addOtherPrePrepare(_pp))
	{
		fprintf(stderr, "orderbasenode::processClean(preprepare) says : adding the preprepare should *not* have failed\n");
		exit(151);
	}
	
	//1 construct the next batch from the preprepare
	boost::shared_ptr<BFT::messages::NextBatch> nb = cert.get()->getNextBatch(baseNode, currentView);
	
	baseNode->authenticateExecMacArrayMessage(nb); //TODO needed? not already authenticated by getNextBatch()?
	
	//2 add the next batch to the order state
	addNextBatch(nb);
	
	//only need to prepare and commit on checkpoint interval
	if(cp)
	{
		//3 construct the prepare
//		fprintf(stderr,"orderbasenode::process(preprepare) says : constructing a prepare for seqno %d\n",seqno);
		boost::shared_ptr<messages::Prepare> p (
			new messages::Prepare(_pp.get()->getView(), _pp.get()->getSeqNo(), 
				cert.get()->getHistory(), baseNode.get()->getMyOrderIndex()) );
		baseNode.get()->authenticateOrderMacArrayMessage(p);
		 
		 //4 add it to the order state
		cert.get()->addMyPrepare(p);
//		fprintf(stderr, "@ Orderbasenode::processclean(preprepare) adding my prepare %d - %d\n",seqno,p.get()->getSender());
		 
		 //5 send it out
		 int p_bytes_size;
		 boost::shared_ptr<char> p_bytes = p.get()->getBytes(p_bytes_size);
		 baseNode.get()->sendToAllOrder(p_bytes, p_bytes_size);
// 		 fprintf(stderr, "@ Orderbasenode::processclean(preprepare) sending prepare %d - %d\n",seqno,p.get()->getSender());

		 if (cert.get()->isPrepared())
		 {
		 	actOnPrepared(p.get()->getSeqNo());
		 	
		 }
		 
		 //6 flush the cached prepares and commits
		 
		 std::vector<boost::shared_ptr<messages::Prepare> > pcache = cert.get()->getPrepareCache();
		 std::vector<boost::shared_ptr<messages::Commit> > ccache = cert.get()->getCommitCache();
		 
		 for(unsigned int i = 0; i < pcache.size(); i++)
		 {
		 	if( p.get()->matches(pcache.at(i)) )
		 	{
		 		process(pcache.at(i));
		 	}
		 	
		 	if( p.get()->matches(ccache.at(i)) )
		 	{
		 		process(ccache.at(i));
		 	}	
		 }
		
	}
	else
	{
		//Not a cp interval, rely on speculation!
//		fprintf(stderr, "@ sending nextbatch for seqno %d\n",nb.get()->getSeqNo());
		writeAndSend(nb);
		
		//try to process a cached preprepare if there is one
		boost::shared_ptr<messages::PrePrepare> pp2 = 
			certificates.at(stateIndex(seqno+1)).at(certificateIndex(seqno+1)).get()->getCachedPrePrepare();
		
		if (pp2.use_count() > 0)
		{
			process(pp2);
		}
		 //6 flush the cached prepares 
		 
		 std::vector<boost::shared_ptr<messages::Prepare> > pcache = cert.get()->getPrepareCache();
		 
		 for(unsigned int i = 0; i < pcache.size(); i++)
		 {
		 	if( pcache.at(i).use_count() > 0 &&
		 	    pcache.at(i).get()->getPrePrepareHash().get()->equals(cert.get()->getHistory()) )
		 	{
		 		process(pcache.at(i));
		 	}
		 }
		
	}
	
	if(     (baseNode.get()->getMyOrderIndex()==1 && getCurrentSequenceNumber()==211 ) 
	     || (baseNode.get()->getMyOrderIndex()==2 && getCurrentSequenceNumber()==221 )
	     /// should be in view 1 after this line
	     || (baseNode.get()->getMyOrderIndex()==3 && getCurrentSequenceNumber()==291 ) 
	     || (baseNode.get()->getMyOrderIndex()==2 && getCurrentSequenceNumber()==293 )
	     /// should be in view 2 after this line
	     || (baseNode.get()->getMyOrderIndex()==0 && getCurrentSequenceNumber()==301 )
	     || (baseNode.get()->getMyOrderIndex()==3 && getCurrentSequenceNumber()==331 )
	     /// should be in view 3 after this line)
     	     || (baseNode.get()->getMyOrderIndex()==0 && getCurrentSequenceNumber()==541 )
	     || (baseNode.get()->getMyOrderIndex()==1 && getCurrentSequenceNumber()==562 )
	     /// should be in view 4 after this line)
     	     || (baseNode.get()->getMyOrderIndex()==2 && getCurrentSequenceNumber()==698 )
	     || (baseNode.get()->getMyOrderIndex()==1 && getCurrentSequenceNumber()==722 )
	)	   	     

	{

		fprintf(stderr, "orderbasenode::process(preprepare) says: should have starting a view change\n");
		startViewChange(currentView+1);
		return;
	}
	
}

void OrderBaseNode::addNextBatch(boost::shared_ptr<BFT::messages::NextBatch> _nb)
{	
	uint32_t lasttime = _nb.get()->getNonDeterminism().get()->getTime();
  	workingState.get()->setCurrentTime(lasttime);

	uint32_t currenttime= millisecondTime();
	boost::shared_ptr<std::vector< boost::shared_ptr<BFT::messages::Entry> > > entries = 
		_nb.get()->getCommands().get()->getEntries();
	//TODO: iterating through the request in the batch, again... to optimize that, put the code within
	// addNextBatch method of CheckPointState
	for (unsigned int i = 0 ; i< entries.get()->size(); i++)
	{
		lastClientRetrans.at(entries.get()->at(i).get()->getClient())=currenttime;
	}
	
	workingState.get()->addNextBatch(_nb.get()->getCommands(), _nb.get()->getSeqNo(), _nb.get()->getHistory() );
}

bool OrderBaseNode::tryRetransmit(boost::shared_ptr<BFT::messages::RequestCore> _rc)
{
	if( _rc.get()->getRequestId() > lastRidForClient.at(_rc.get()->getSendingClient()))
	{
		lastRidForClient.at(_rc.get()->getSendingClient())=_rc.get()->getRequestId();
		return false;
	}

	uint32_t lastOrdered = workingState.get()->getLastOrdered( _rc.get()->getSendingClient() );
	
	if (lastOrdered >= _rc.get()->getRequestId() )
	{
		//rate limiting logic
		uint32_t client = _rc.get()->getSendingClient();
		if(millisecondTime() - lastClientRetrans.at(client) < workingState.get()->getRetransmitDelay(client))
		{
//			fprintf(stderr, "ignoring retransmission from client %d, too soon\n",client);
			return true;
		}
		workingState.get()->updateRetransmitDelay(client);
	
//		fprintf(stderr,"received a valid retransmisison from client %d - rid %d - last ordered %d\n"
//			,client,_rc.get()->getRequestId(),lastOrdered );
		
		boost::shared_ptr<BFT::messages::Retransmit> ret(
			new BFT::messages::Retransmit(client, 
						      workingState.get()->getCurrentSequenceNumber(), 
						      baseNode.get()->getMyOrderIndex() ) );
		baseNode.get()->authenticateExecMacArrayMessage(ret);
		lastClientRetrans.at(client) = millisecondTime();

		int ret_size;
		boost::shared_ptr<char> ret_bytes = ret.get()->getBytes(ret_size);
		baseNode.get()->sendToAllExec(ret_bytes, ret_size);
		/*
		TODO (or maybe not)
		skipping the preprepare retransmission	
		
		(14:23:17) allen clement: basically, we want to send the retransmit message (the first bit) but we dont want to 'escalate' to send the prepare messages (yet).
		(14:23:43) Turoldo: ack
		(14:24:26) allen clement: in the long term, need to change the code so only the primary sends the prepare, dont want the non-primaries to do it spontaneously.	
		*/

//		if (/*lastOrdered > mp &&*/ !changingView /*&& amIPrimary(currentView)*/)
//		{
//			uint32_t seqno = getCurrentSequenceNumber() -1;
//			boost::shared_ptr<statemanagement::Certificate> cert = 
//				certificates.at(stateIndex(seqno)).at(certificateIndex(seqno));
//			// If I havent prepared yet, then prepare
//		    	
//			if (cert.get()->getPrepare().use_count() == 0)
//			{
//				fprintf(stderr, "QQQQQQQQQQQQQQQ\n");
//		    		boost::shared_ptr<messages::Prepare> p (
//		    			new messages::Prepare(currentView, 
//					    	    	      cert.get()->getSeqNo(),
//					    	    	      cert.get()->getHistory(),
//					            	      baseNode.get()->getMyOrderIndex()) );
//				cert.get()->addMyPrepare(p);
//		    		baseNode.get()->authenticateOrderMacArrayMessage(p);
//	    			int p_bytes_size;
//	    			boost::shared_ptr<char> p_bytes = p.get()->getBytes();
//				baseNode.get()->sendToAllOrder(p_bytes, p_bytes_size);
//			}
//		}


		if(amIPrimary(currentView))
		{

			//send a prepare for the highest preprepared
			uint32_t last_ordered = workingState.get()->getCurrentSequenceNumber()-1;
			boost::shared_ptr<BFT::messages::HistoryDigest> hd = 
				certificates.at(stateIndex(last_ordered)).at(certificateIndex(last_ordered)).get()->getHistory();

			uint32_t seqno = certificates.at(stateIndex(last_ordered)).at(certificateIndex(last_ordered)).get()->getSeqNo();
			if(certificates.at(stateIndex(last_ordered)).at(certificateIndex(last_ordered)).get()->getPrepare().use_count()==0)
			{
				fprintf(stderr, "------ issuing a prepare due to client retransmission. view %d, seqno %d\n",
					currentView, seqno);
				boost::shared_ptr<messages::Prepare> p (
					new messages::Prepare(currentView, 
							      seqno,
							      hd,
							      baseNode.get()->getMyOrderIndex() ));
				
				certificates.at(stateIndex(last_ordered)).at(certificateIndex(last_ordered)).get()->addMyPrepare(p);			    
				baseNode.get()->authenticateOrderMacArrayMessage(p);    
				int p_bytes_size;
				boost::shared_ptr<char> p_bytes = p.get()->getBytes(p_bytes_size);
				baseNode.get()->sendToAllOrder(p_bytes, p_bytes_size);			      
//				fprintf(stderr, "------ issued a prepare for seqno %d due to client retransmission\n",seqno);
//				handle(p_bytes, p_bytes_size);
			}
		}	




		
		return true;
	}
	else
	{
		return false;
	}
}

bool OrderBaseNode::canCreateBatch()
{
	//can't create batch if we are changing view
	if (changingView)
		return false;
	
	if(!stateSnapshots.at(baseIndex).get()->isCommitted())
	{
		if (getCurrentSequenceNumber() == 0)
		{
			//fprintf(stderr, "obn::cancreatebatch says: false, because state not committed and currentsequence==0");
			return false;
		}
		else
		{
			fprintf(stderr, "base should always be stable\n");
			exit(118);
		}
	}
	
	uint32_t seqNo = getCurrentSequenceNumber();
	
	if( seqNo % BFT::order::Parameters::checkPointInterval == 0 &&
	    seqNo > 0 &&
	    !certificates.at(stateIndex(seqNo -1)).at(certificateIndex(seqNo-1)).get()->isCommitted() )
	{
		//fprintf(stderr,"OrderBaseNode :: canCreateBatch says : false 1\n");
		return false;
	}
	
	int baseplusone = (baseIndex+1) % stateSnapshots.size();
	int baseminusone = (baseIndex + stateSnapshots.size() -1) % stateSnapshots.size();
	
	if( currentIndex == baseminusone &&
	    (seqNo + 1) % BFT::order::Parameters::checkPointInterval == 0)
	{
	    	if(!stateSnapshots.at(baseplusone).get()->isStable())
    		{
			fprintf(stderr,"OrderBaseNode :: canCreateBatch says : false 2\n");
			return false;
		}
	}
	
	return true;
}

int OrderBaseNode::stateIndex(uint32_t _seqNo)
{
	uint32_t base = getBaseSequenceNumber();
	base = _seqNo - base;
	base = base / BFT::order::Parameters::checkPointInterval;
	return (int) (baseIndex + base)%BFT::order::Parameters::maxPeriods;
}


void OrderBaseNode::start()
{
	boost::shared_ptr<BFT::messages::RequestCP> rcp (
		new BFT::messages::RequestCP(getBaseSequenceNumber(), baseNode.get()->getMyOrderIndex() ) );
		
	baseNode.get()->authenticateExecMacArrayMessage(rcp);
	int bytes_size;
	boost::shared_ptr<char> bytes = rcp.get()->getBytes(bytes_size);
	baseNode.get()->sendToAllExec(bytes, bytes_size);
}

void OrderBaseNode::writeAndSend(boost::shared_ptr<BFT::messages::NextBatch> _nb)
{
	if(BFT::order::Parameters::doLogging)
	{
		//TODO - implement logging related stuff
	}
	else
	{
		if (_nb.use_count() == 0)
		{	
			fprintf(stderr, "orderbasenode::writeandsend , invalid nextbatch\n");
			exit(150);
		}
		int nbsize;
		boost::shared_ptr<char> nbbytes = _nb.get()->getBytes(nbsize);
//		fprintf(stderr, "orderbasenode::writeandsend says : sending a next batch message\n");
		baseNode.get()->sendToAllExec(nbbytes,nbsize);
	}

}

/**
Processes a prepared sequence number

(1)  assert if not prepared
(2) create a new commit message
(3)  add it to the state
(4) send it out
(5) if that commits the request, act on the commitment
 **/
void OrderBaseNode::actOnPrepared(uint32_t _seqno){
//	fprintf(stderr,"\t\tactonprepared %d\n",_seqno);
	boost::shared_ptr<statemanagement::Certificate> cert = 
		certificates.at(stateIndex(_seqno)).at(certificateIndex(_seqno));
	// (1)  assert that the sequence number is in fact prepared
	if (!cert.get()->isPrepared())
	{
		fprintf(stderr, "orderbasenode::actonprepared says : it is not prepared!\n");
		exit(129);
	}
	if ((int32_t)_seqno > maxPrepared)
		maxPrepared = (int32_t)_seqno;

	// (2)  create the new commit
	boost::shared_ptr<messages::Commit> c ( 
		new messages::Commit(currentView, 
				     _seqno, 
				     cert.get()->getHistory(),
				     baseNode.get()->getMyOrderIndex()) );
				     
	baseNode.get()->authenticateOrderMacArrayMessage(c);
	// (3) add it to the state
	if (!cert.get()->addMyCommit(c))
	{
		fprintf(stderr, "orderbasenode::actonprepared says : should not have failed!\n");
		exit(130);

	}
	// (4) send it out
	int csize;
	boost::shared_ptr<char> cbytes = c.get()->getBytes(csize);
	baseNode.get()->sendToAllOrder(cbytes,csize);
	// (5) check to see if committed
	if (cert.get()->isCommitted())
		actOnCommitted(_seqno);

	//TODO that is a LOT of loops. possible optimizations? (keepping a lastprepared, or something?)
	if (certificateIndex(_seqno) != 0)
		for (int i = certificateIndex(_seqno-1); 
			i >= 0 && !certificates.at(stateIndex(_seqno)).at(i).get()->isPrepared() ; 
			i--)
		{
			certificates.at(stateIndex(_seqno)).at(i).get()->forcePrepared();
		}
}

/**
Act on committed requests. 

(1)  fetch the nextbatch message for this sequence number
(2)  send it out
(3)  if this is the last seqno before the next checkpointinterval
(3b1)  create a new working state
(3bc)  make the checkpoint stable if it has the app cp token
(4)  process cached preprepare if it exists
(5)  if a checkpoint is complete, mark previous entries as committed
 **/

void OrderBaseNode::actOnCommitted(uint32_t _seqno) 
{
	fprintf(stderr, "orderbasenode::actoncommit says : acting on Committed %d\n",_seqno);
	
	boost::shared_ptr<statemanagement::Certificate> cert =
		certificates.at(stateIndex(_seqno)).at(certificateIndex(_seqno));
	
	if (!cert.get()->isCommitted())
	{
		fprintf(stderr, "orderbasenode::actoncommitted says : certificate not committed!\n");
		exit(131);
	}
	
	if ((int32_t)_seqno > maxCommitted)
		maxCommitted = _seqno;
		
	//fetch the next batch for this seqno
	boost::shared_ptr<BFT::messages::NextBatch> nb = 
		cert.get()->getNextBatch(baseNode, currentView);
	
	baseNode.get()->authenticateExecMacArrayMessage(nb);
	
//	// testing the nb correctness
//	int nb_bytes_size;
//	boost::shared_ptr<char> nb_bytes = nb.get()->getBytes(nb_bytes_size);
//	boost::shared_ptr<BFT::messages::VerifiedMessageBase> test_vmb (
//		new BFT::messages::VerifiedMessageBase(nb_bytes, nb_bytes_size) );
//	nb_bytes = test_vmb.get()->getBytes(nb_bytes_size);	
//	boost::shared_ptr<BFT::messages::NextBatch> test_nb (
//		new BFT::messages::NextBatch(nb_bytes, nb_bytes_size));
	
	//send it out
	writeAndSend(nb);
	
//	nb.get()->printInfo();
	
	//if this a checkpointintervalth request?
	if ( (_seqno+1) % BFT::order::Parameters::checkPointInterval == 0)
	{
		stateSnapshots.at(currentIndex).get()->commit();
		boost::shared_ptr<statemanagement::CheckPointState> cps = workingState;
		//create the new state
		newWorkingState();
		//make the old cp stable if it has the application token
		if(cps.get()->hasExecCP())
		{
			makeCPStable(cps);
		}
		else
		{
			fprintf(stderr, "orderBaseNode::actoncommitted says: NOT making stable, no execCPtoken yet\n");
		}
		
	}
	
	
	// (3c)  commit all requests in the current set of commit certificates
	if (certificateIndex(_seqno) != 0)
	{
	    	for (int i = certificateIndex(_seqno-1); 
		 	i >= 0 && !certificates.at(stateIndex(_seqno)).at(i).get()->isCommitted(); 
		 	i--)
		{
			certificates.at(stateIndex(_seqno)).at(i).get()->forceCommitted();
	    	}
	}
	// (4) try to process a cached preprepare if it exists
	boost::shared_ptr<messages::PrePrepare> pp = 
	    certificates.at(stateIndex(_seqno+1)).at(certificateIndex(_seqno+1)).get()->getCachedPrePrepare();
	if (pp.use_count() != 0)
	{
		process(pp);
	}
	
//	fprintf(stderr, "orderbasenode::actoncommit says : ended acting on Committed %d\n",_seqno);
}

/** move on to a new working state **/
void OrderBaseNode::newWorkingState()
{
//	fprintf(stderr, "orderbasenode::newworkingstate says : creating new workingstate\n");
	if (getCurrentSequenceNumber() % BFT::order::Parameters::checkPointInterval != 0)
	{
		fprintf(stderr, 
			"orderbasenode::newworkingstate says : trying to get a new working state prematurely,%d\n"
			,getCurrentSequenceNumber());
		exit(131);
	}
	
	int baseplusone = (baseIndex + 1) %BFT::order::Parameters::maxPeriods;
	if (stateSnapshots[baseplusone].use_count() != 0)
	{
		if (currentIndex != baseIndex &&
		    !stateSnapshots.at(baseplusone).get()->isCommitted())
		{
			fprintf(stderr,"orderbasenode::newworkingstate says : WTF?\n");
			exit(135); //TODO check if the exit is necessary
		}
	} 
	
	//garbage collect the first index
	if (currentIndex != baseIndex)
		garbageCollect(stateSnapshots.at(baseplusone).get()->getCurrentSequenceNumber() );

//	fprintf(stderr, "orderbasenode::newworkingstate says : first index garbagecollected\n");
	
	if( (currentIndex + 1) % BFT::order::Parameters::maxPeriods == baseIndex )
	{
		fprintf(stderr, "orderbasenode::newworkingstate says : need to garbage collect first\n");
		exit(133);
	}
	
	currentIndex = (currentIndex+1)%BFT::order::Parameters::maxPeriods;
	stateSnapshots[currentIndex] = workingState;
	workingState.reset( new statemanagement::CheckPointState (workingState));
	//TODO implement NewLog
//		newLog(workingState.getCurrentSequenceNumber() + "_" + getMyOrderIndex() + "_");
	for(unsigned int i =0; i< futurePPs.size(); i++)
	{
		if (futurePPs.at(i).use_count() != 0)
		{
			process(futurePPs.at(i));
		}
		futurePPs.at(i).reset();
		for (unsigned int j = 0; j< futurePs.at(i).size() ; j++)
		{
			if (futurePs.at(i).at(j).use_count() != 0)
			{
				process(futurePs.at(i).at(j));
			}
			
			if (futureCs.at(i).at(j).use_count() != 0)
			{
				process(futureCs.at(i).at(j));
			}
			
			futurePs.at(i).at(j).reset();
			futureCs.at(i).at(j).reset();
		}
	}
//	fprintf(stderr, "orderbasenode::newworkingstate says : newworkingstate created successfully\n");
}



/**
garbage collect checkpoints less than seqno
 **/
 
void OrderBaseNode::garbageCollect(uint32_t _seqno)
{

	if (_seqno >= getCurrentSequenceNumber())
	{
		fprintf(stderr, "orderbasenode::garbagecollect says : trying to garbage collect the future\n");
		exit(134);
	}
	
	if (_seqno % BFT::order::Parameters::checkPointInterval != 0)
	{
		garbageCollect(_seqno - (_seqno % BFT::order::Parameters::checkPointInterval));
		return;
	}
	
	while (getBaseSequenceNumber() < _seqno &&
		baseIndex != currentIndex)
	{
//		fprintf(stderr,"orderbasenode::garbagecollect says: -- 1 --\n");
		int baseplusone = (baseIndex + 1)%BFT::order::Parameters::maxPeriods;
//		fprintf(stderr,"orderbasenode::garbagecollect says: -- 2 --\n");
		if(!stateSnapshots.at(baseplusone).get()->isCommitted())
		{
			fprintf(stderr,"orderbasenode::garbagecollect says: garbage collecting when the next snapshot is not stable\n");
			exit(134);
		}
//		fprintf(stderr,"orderbasenode::garbagecollect says: -- 3 --\n");
		if(!stateSnapshots.at(baseIndex).get()->isCommitted())
		{
			fprintf(stderr,"orderbasenode::garbagecollect says: garbage collecting something that is not stable\n");
			exit(134);
		}
//		fprintf(stderr,"orderbasenode::garbagecollect says: -- 4 --\n");
		for(unsigned int i=0; i<certificates.at(baseIndex).size(); i++)
		{
//			fprintf(stderr,"orderbasenode::garbagecollect says: -- 5 - %d --\n",i);		
			certificates.at(baseIndex).at(i).reset( 
				new statemanagement::Certificate(BFT::Parameters::getOrderCount(),
								((int)i == BFT::order::Parameters::checkPointInterval-1)?true:false)
							      );
		}
		
		boost::shared_ptr<statemanagement::CheckPointState> cps = stateSnapshots.at(baseIndex);
		int cptokenbytessize;
		boost::shared_ptr<char> cptokenbytes = cps.get()->getExecCPToken(cptokenbytessize);
//		fprintf(stderr,"orderbasenode::garbagecollect says: -- 6 --\n");
		if(cps.use_count() == 0 || cptokenbytes.use_count() == 0)
		{
			fprintf(stderr, "All hands -- ABANDON SHIP! \n");
			exit(134);
		}
		boost::shared_ptr<BFT::messages::ReleaseCP> rcp (
			new BFT::messages::ReleaseCP(cptokenbytes, cptokenbytessize,
						     cps.get()->getCurrentSequenceNumber(),
						     baseNode.get()->getMyOrderIndex()));
//		fprintf(stderr,"orderbasenode::garbagecollect says: -- 7 --\n");						     
		baseNode.get()->authenticateExecMacArrayMessage(rcp);
		int rcpbytessize;
		boost::shared_ptr<char> rcpbytes = rcp.get()->getBytes(rcpbytessize);
		baseNode->sendToAllExec(rcpbytes,rcpbytessize);
//		rcp.get()->printInfo();
		
		//TODO implement logging
		/*
		gcLog(cps.getCurrentSequenceNumber() + "_" + getMyOrderIndex() + "_");
		File cpfile = new File(cps.getCurrentSequenceNumber() + "_CP.LOG");
		if(!cpfile.exists()) {
			cpfile = new File(cps.getCurrentSequenceNumber() + "_" + getMyOrderIndex() + "_CP.LOG");
		}
		if(cpfile.exists()) {
			cpfile.delete();
			//BFT.//Debug.println("DELETED CP LOG: " + cpfile.getName());
		}		

		*/
		stateSnapshots.at(baseIndex).reset();
		baseIndex = (baseIndex+1) % BFT::order::Parameters::maxPeriods;
	}
//	fprintf(stderr, "orderbasenode::garbagecollect says : garbage collectedfor seqno %d\n",_seqno);	
}


uint32_t OrderBaseNode::millisecondTime()
{
  //clock_gettime(CLOCK_REALTIME, &time);
  gettimeofday(&time, NULL);
	uint32_t msec = time.tv_sec*1000 + time.tv_usec/1000;
	return msec;
}
